From 54ac19454babd61f9c2a02096fa64c054ed873a0 Mon Sep 17 00:00:00 2001
From: GeorgiaPreda <georgiapreda@yahoo.com>
Date: Thu, 22 Oct 2026 22:45:35 -0400
Subject: [PATCH 1/2] Fix worker Prometheus collection during startup

---
 distributed/http/worker/prometheus/core.py | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/distributed/http/worker/prometheus/core.py b/distributed/http/worker/prometheus/core.py
index 3b3084e4f3..009e1138eb 100644
--- a/distributed/http/worker/prometheus/core.py
+++ b/distributed/http/worker/prometheus/core.py
@@ -171,7 +171,7 @@ def collect_crick(self) -> Iterator[Metric]:
         # The following metrics will export NaN, if the corresponding digests are None
         if not self.crick_available:
             return
-        assert self.server.digests
+        assert self.server.digests is not None
 
         yield GaugeMetricFamily(
             self.build_name("tick_duration_median"),
@@ -262,9 +262,11 @@ def __init__(self, *args, dask_server=None, **kwargs):
             PrometheusHandler._collector.server = self.server
             return
 
-        PrometheusHandler._collector = WorkerMetricCollector(self.server)
-        # Register collector
-        prometheus_client.REGISTRY.register(PrometheusHandler._collector)
+        collector = WorkerMetricCollector(self.server)
+        # Registration collects metrics and may fail. Only cache a registered
+        # collector so a later request can retry after a failure.
+        prometheus_client.REGISTRY.register(collector)
+        PrometheusHandler._collector = collector
 
     def get(self):
         self.write(prometheus_client.generate_latest())

From ff3fde229afec06ca1f9afcba4cddf085432931e Mon Sep 17 00:00:00 2001
From: GeorgiaPreda <georgiapreda@yahoo.com>
Date: Thu, 22 Oct 2026 22:47:18 -0400
Subject: [PATCH 2/2] Test early worker metrics scrapes and registration retry

---
 .../http/worker/tests/test_worker_http.py     | 65 +++++++++++++++++++
 1 file changed, 65 insertions(+)

diff --git a/distributed/http/worker/tests/test_worker_http.py b/distributed/http/worker/tests/test_worker_http.py
index 4375e8e15f..f935ee1ae5 100644
--- a/distributed/http/worker/tests/test_worker_http.py
+++ b/distributed/http/worker/tests/test_worker_http.py
@@ -15,6 +15,71 @@
 )
 
 
+@gen_cluster(nthreads=[])
+async def test_prometheus_before_worker_registered(s, monkeypatch):
+    prometheus_client = pytest.importorskip("prometheus_client")
+    pytest.importorskip("crick")
+    from distributed.http.worker.prometheus.core import PrometheusHandler
+
+    registry = prometheus_client.CollectorRegistry(auto_describe=True)
+    generate_latest = prometheus_client.generate_latest
+    monkeypatch.setattr(prometheus_client, "REGISTRY", registry)
+    monkeypatch.setattr(
+        prometheus_client, "generate_latest", lambda: generate_latest(registry)
+    )
+    monkeypatch.setattr(PrometheusHandler, "_collector", None)
+
+    class StartupWorker(Worker):
+        async def _register_with_scheduler(self):
+            # HTTP is already listening, but periodic callbacks have not populated
+            # the worker's digests yet. Prometheus may discover this endpoint now.
+            assert self.digests is not None
+            assert not self.digests
+            metrics = await fetch_metrics_sample_names(
+                self.http_server.port, prefix="dask_worker_"
+            )
+            assert "dask_worker_threads" in metrics
+            assert "dask_worker_tick_duration_median_seconds" in metrics
+            await super()._register_with_scheduler()
+
+    async with StartupWorker(s.address, dashboard_address=":0") as worker:
+        for _ in range(2):
+            metrics = await fetch_metrics_sample_names(
+                worker.http_server.port, prefix="dask_worker_"
+            )
+            assert "dask_worker_threads" in metrics
+
+
+@gen_cluster(nthreads=[("127.0.0.1", 1)])
+async def test_prometheus_registration_retry(s, a, monkeypatch):
+    prometheus_client = pytest.importorskip("prometheus_client")
+    from tornado.simple_httpclient import HTTPStreamClosedError
+
+    from distributed.http.worker.prometheus.core import PrometheusHandler
+
+    registry = prometheus_client.CollectorRegistry(auto_describe=True)
+    generate_latest = prometheus_client.generate_latest
+    monkeypatch.setattr(prometheus_client, "REGISTRY", registry)
+    monkeypatch.setattr(
+        prometheus_client, "generate_latest", lambda: generate_latest(registry)
+    )
+    monkeypatch.setattr(PrometheusHandler, "_collector", None)
+
+    with (
+        mock.patch.object(
+            registry, "register", side_effect=RuntimeError("test failure")
+        ),
+        pytest.raises(HTTPStreamClosedError),
+    ):
+        await fetch_metrics_sample_names(a.http_server.port, prefix="dask_worker_")
+
+    for _ in range(2):
+        metrics = await fetch_metrics_sample_names(
+            a.http_server.port, prefix="dask_worker_"
+        )
+        assert "dask_worker_threads" in metrics
+
+
 @gen_cluster(
     client=True,
     nthreads=[("127.0.0.1", 1)],