From 0cad98d8b125c371a09f82c3bc2db2116db6c1d7 Mon Sep 17 00:00:00 2001 From: Gabriel Sailer Date: Fri, 21 Jun 2019 10:29:12 +0200 Subject: [PATCH 1/6] Expose tasks prometheus metric at scheduler --- distributed/dashboard/scheduler_html.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/distributed/dashboard/scheduler_html.py b/distributed/dashboard/scheduler_html.py index 08829241d47..333de2da66a 100644 --- a/distributed/dashboard/scheduler_html.py +++ b/distributed/dashboard/scheduler_html.py @@ -192,7 +192,11 @@ def collect(self): "Number of clients.", value=len(self.server.clients), ) - + yield GaugeMetricFamily( + "dask_scheduler_tasks", + "Number of tasks", + value=len(self.server.tasks) + ) class PrometheusHandler(RequestHandler): _initialized = False From fe2c0e07376d54837c730809e835d4336060e1d3 Mon Sep 17 00:00:00 2001 From: Gabriel Sailer Date: Fri, 28 Jun 2019 16:42:48 +0200 Subject: [PATCH 2/6] Add basic task metrics to worker Number of tasks in states and number of threads are exposed on the workers /metrics endpoints. --- distributed/dashboard/worker_html.py | 44 ++++++++++++++++++++-------- 1 file changed, 31 insertions(+), 13 deletions(-) diff --git a/distributed/dashboard/worker_html.py b/distributed/dashboard/worker_html.py index e1ae50f3afc..54500f0a1a3 100644 --- a/distributed/dashboard/worker_html.py +++ b/distributed/dashboard/worker_html.py @@ -3,21 +3,39 @@ class _PrometheusCollector(object): def __init__(self, server, prometheus_client): - self.server = server + self.worker = server 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 + + yield GaugeMetricFamily( + "dask_worker_tasks_stored", + "Number of tasks stored", + value=len(self.worker.data), + ) + yield GaugeMetricFamily( + "dask_worker_tasks_ready", + "Number of tasks ready", + value=len(self.worker.ready), + ) + yield GaugeMetricFamily( + "dask_worker_tasks_waiting", + "Number of tasks waiting", + value=len(self.worker.waiting_for_data), + ) + yield GaugeMetricFamily( + "dask_worker_connections", + "Number of task connections", + value=len(self.worker.in_flight_workers), + ) + yield GaugeMetricFamily( + "dask_worker_tasks_serving", + "Number of tasks serving", + value=len(self.worker._comms), + ) + yield GaugeMetricFamily( + "dask_worker_nthreads", "Number of threads", value=self.worker.nthreads + ) class PrometheusHandler(RequestHandler): From 4a0bd2ad02e193bd6eeea45f8cef3c08a304152e Mon Sep 17 00:00:00 2001 From: Gabriel Sailer Date: Thu, 11 Jul 2019 19:04:44 +0200 Subject: [PATCH 3/6] Add worker metrics and reformat tasks --- distributed/dashboard/scheduler_html.py | 16 ++++-- distributed/dashboard/worker_html.py | 76 +++++++++++++++++-------- 2 files changed, 64 insertions(+), 28 deletions(-) diff --git a/distributed/dashboard/scheduler_html.py b/distributed/dashboard/scheduler_html.py index 333de2da66a..3f119a929b9 100644 --- a/distributed/dashboard/scheduler_html.py +++ b/distributed/dashboard/scheduler_html.py @@ -184,19 +184,25 @@ 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_tasks", - "Number of tasks", - value=len(self.server.tasks) + "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): _initialized = False diff --git a/distributed/dashboard/worker_html.py b/distributed/dashboard/worker_html.py index 54500f0a1a3..5678589c553 100644 --- a/distributed/dashboard/worker_html.py +++ b/distributed/dashboard/worker_html.py @@ -1,42 +1,74 @@ +import logging from .utils import RequestHandler, redirect class _PrometheusCollector(object): - def __init__(self, server, prometheus_client): + def __init__(self, server): self.worker = server + self.logger = logging.getLogger("distributed.dask_worker") + self.crick_available = True + try: + import crick + + # usage of import necessary to comply with flake8 F401 + crick_version = crick._version + except ImportError: + self.crick_available = False + self.logger.info("Digest-based metrics require crick to be installed") def collect(self): from prometheus_client.core import GaugeMetricFamily - yield GaugeMetricFamily( - "dask_worker_tasks_stored", - "Number of tasks stored", - value=len(self.worker.data), - ) - yield GaugeMetricFamily( - "dask_worker_tasks_ready", - "Number of tasks ready", - value=len(self.worker.ready), - ) - yield GaugeMetricFamily( - "dask_worker_tasks_waiting", - "Number of tasks waiting", - value=len(self.worker.waiting_for_data), + 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", + "Number of task connections to other workers.", value=len(self.worker.in_flight_workers), ) + yield GaugeMetricFamily( - "dask_worker_tasks_serving", - "Number of tasks serving", - value=len(self.worker._comms), + "dask_worker_threads", + "Number of worker threads.", + value=self.worker.nthreads, ) + yield GaugeMetricFamily( - "dask_worker_nthreads", "Number of threads", value=self.worker.nthreads + "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): _initialized = False @@ -49,9 +81,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 From f4d7bf202de02942d1b7c7cfb872b720dfa6f68a Mon Sep 17 00:00:00 2001 From: Gabriel Sailer Date: Thu, 11 Jul 2019 19:07:02 +0200 Subject: [PATCH 4/6] Change prometheus worker test to check for a specific metric --- distributed/dashboard/tests/test_worker_bokeh_html.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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}}) From eb8e591de1f8a6a5572ae0d7491b36bef9e204ae Mon Sep 17 00:00:00 2001 From: Gabriel Sailer Date: Fri, 12 Jul 2019 14:46:24 +0200 Subject: [PATCH 5/6] Change unused import handling to linter ignore notation --- distributed/dashboard/worker_html.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/distributed/dashboard/worker_html.py b/distributed/dashboard/worker_html.py index 5678589c553..9a317b66099 100644 --- a/distributed/dashboard/worker_html.py +++ b/distributed/dashboard/worker_html.py @@ -8,10 +8,7 @@ def __init__(self, server): self.logger = logging.getLogger("distributed.dask_worker") self.crick_available = True try: - import crick - - # usage of import necessary to comply with flake8 F401 - crick_version = crick._version + import crick # noqa: F401 except ImportError: self.crick_available = False self.logger.info("Digest-based metrics require crick to be installed") From b044bc4de7db9e7dd6c347764a3d6a6809dae692 Mon Sep 17 00:00:00 2001 From: Gabriel Sailer Date: Tue, 16 Jul 2019 10:42:13 +0200 Subject: [PATCH 6/6] Change log mesage in case of missing crick --- distributed/dashboard/worker_html.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/distributed/dashboard/worker_html.py b/distributed/dashboard/worker_html.py index 9a317b66099..27e1f9fe9d2 100644 --- a/distributed/dashboard/worker_html.py +++ b/distributed/dashboard/worker_html.py @@ -11,7 +11,9 @@ def __init__(self, server): import crick # noqa: F401 except ImportError: self.crick_available = False - self.logger.info("Digest-based metrics require crick to be installed") + self.logger.info( + "Not all prometheus metrics available are exported. Digest-based metrics require crick to be installed" + ) def collect(self): from prometheus_client.core import GaugeMetricFamily