diff --git a/distributed/dashboard/scheduler_html.py b/distributed/dashboard/scheduler_html.py index 08829241d47..3f119a929b9 100644 --- a/distributed/dashboard/scheduler_html.py +++ b/distributed/dashboard/scheduler_html.py @@ -184,14 +184,24 @@ def collect(self): yield GaugeMetricFamily( "dask_scheduler_workers", - "Number of workers.", + "Number of workers connected.", value=len(self.server.workers), ) yield GaugeMetricFamily( "dask_scheduler_clients", - "Number of clients.", + "Number of clients connected.", value=len(self.server.clients), ) + yield GaugeMetricFamily( + "dask_scheduler_received_tasks", + "Number of tasks received at scheduler", + value=len(self.server.tasks), + ) + yield GaugeMetricFamily( + "dask_scheduler_unrunnable_tasks", + "Number of unrunnable tasks at scheduler", + value=len(self.server.unrunnable), + ) class PrometheusHandler(RequestHandler): diff --git a/distributed/dashboard/tests/test_worker_bokeh_html.py b/distributed/dashboard/tests/test_worker_bokeh_html.py index 99916b3fdc7..7a4d70a037c 100644 --- a/distributed/dashboard/tests/test_worker_bokeh_html.py +++ b/distributed/dashboard/tests/test_worker_bokeh_html.py @@ -25,7 +25,7 @@ def test_prometheus(c, s, a, b): txt = response.body.decode("utf8") families = {familiy.name for familiy in text_string_to_metric_families(txt)} - assert len(families) > 0 + assert "dask_worker_latency_seconds" in families @gen_cluster(client=True, worker_kwargs={"services": {("dashboard", 0): BokehWorker}}) diff --git a/distributed/dashboard/worker_html.py b/distributed/dashboard/worker_html.py index e1ae50f3afc..27e1f9fe9d2 100644 --- a/distributed/dashboard/worker_html.py +++ b/distributed/dashboard/worker_html.py @@ -1,23 +1,72 @@ +import logging from .utils import RequestHandler, redirect class _PrometheusCollector(object): - def __init__(self, server, prometheus_client): - self.server = server + def __init__(self, server): + self.worker = server + self.logger = logging.getLogger("distributed.dask_worker") + self.crick_available = True + try: + import crick # noqa: F401 + except ImportError: + self.crick_available = False + self.logger.info( + "Not all prometheus metrics available are exported. Digest-based metrics require crick to be installed" + ) def collect(self): - # add your metrics here: - # - # 1. remove the following lines - while False: - yield None - # - # 2. yield your metrics - # yield prometheus_client.core.GaugeMetricFamily( - # 'dask_worker_connections', - # 'Number of connections currently open.', - # value=???, - # ) + from prometheus_client.core import GaugeMetricFamily + + tasks = GaugeMetricFamily( + "dask_worker_tasks", "Number of tasks at worker.", labels=["state"] + ) + tasks.add_metric(["stored"], len(self.worker.data)) + tasks.add_metric(["ready"], len(self.worker.ready)) + tasks.add_metric(["waiting"], len(self.worker.waiting_for_data)) + tasks.add_metric(["serving"], len(self.worker._comms)) + yield tasks + + yield GaugeMetricFamily( + "dask_worker_connections", + "Number of task connections to other workers.", + value=len(self.worker.in_flight_workers), + ) + + yield GaugeMetricFamily( + "dask_worker_threads", + "Number of worker threads.", + value=self.worker.nthreads, + ) + + yield GaugeMetricFamily( + "dask_worker_latency_seconds", + "Latency of worker connection.", + value=self.worker.latency, + ) + + # all metrics using digests require crick to be installed + # the following metrics will export NaN, if the corresponding digests are None + if self.crick_available: + yield GaugeMetricFamily( + "dask_worker_tick_duration_median_seconds", + "Median tick duration at worker.", + value=self.worker.digests["tick-duration"].components[1].quantile(50), + ) + + yield GaugeMetricFamily( + "dask_worker_task_duration_median_seconds", + "Median task runtime at worker.", + value=self.worker.digests["task-duration"].components[1].quantile(50), + ) + + yield GaugeMetricFamily( + "dask_worker_transfer_bandwidth_median_bytes", + "Bandwidth for transfer at worker in Bytes.", + value=self.worker.digests["transfer-bandwidth"] + .components[1] + .quantile(50), + ) class PrometheusHandler(RequestHandler): @@ -31,9 +80,7 @@ def __init__(self, *args, **kwargs): if PrometheusHandler._initialized: return - prometheus_client.REGISTRY.register( - _PrometheusCollector(self.server, prometheus_client) - ) + prometheus_client.REGISTRY.register(_PrometheusCollector(self.server)) PrometheusHandler._initialized = True