Extend prometheus metrics endpoint (#2792)

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):

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}})

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