We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add Prometheus counter for `SystemMonitor.last_time` to improve GIL contention metric
Currently, the loose coupling between the system monitor's update interval and the Prometheus scraping interval can cause artifacts like a relative GIL contention > 1 (https://github.com/dask/distributed/pull/7651#issuecomment-1490571845). By exposing the system monitor's update timestamp as a Counter, we would have a synchronized timestamp available in Prometheus to serve as the basis for rate calculations. This should make such artifacts impossible.

cc @ntabris, @gjoseph92, @milesgranger: Thoughts?

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/http/scheduler/prometheus/core.py`
Content:
```
1 from __future__ import annotations
2 
3 from collections.abc import Iterator
4 from time import time
5 
6 import prometheus_client
7 import toolz
8 from prometheus_client.core import CounterMetricFamily, GaugeMetricFamily
9 
10 from distributed.http.prometheus import PrometheusCollector
11 from distributed.http.scheduler.prometheus.semaphore import SemaphoreMetricCollector
12 from distributed.http.scheduler.prometheus.stealing import WorkStealingMetricCollector
13 from distributed.http.utils import RequestHandler
14 from distributed.scheduler import ALL_TASK_STATES, Scheduler
15 
16 
17 class SchedulerMetricCollector(PrometheusCollector):
18     server: Scheduler
19 
20     def __init__(self, server: Scheduler):
21         super().__init__(server)
22         self.subsystem = "scheduler"
23 
24     def collect(self) -> Iterator[GaugeMetricFamily | CounterMetricFamily]:
25         yield GaugeMetricFamily(
26             self.build_name("clients"),
27             "Number of clients connected",
28             value=len([k for k in self.server.clients if k != "fire-and-forget"]),
29         )
30 
31         yield GaugeMetricFamily(
32             self.build_name("desired_workers"),
33             "Number of workers scheduler needs for task graph",
34             value=self.server.adaptive_target(),
35         )
36 
37         worker_states = GaugeMetricFamily(
38             self.build_name("workers"),
39             "Number of workers known by scheduler",
40             labels=["state"],
41         )
42         worker_states.add_metric(["idle"], len(self.server.idle))
43         worker_states.add_metric(
44             ["partially_saturated"],
45             len(self.server.running)
46             - len(self.server.idle)
47             - len(self.server.saturated),
48         )
49         worker_states.add_metric(["saturated"], len(self.server.saturated))
50         worker_states.add_metric(
51             ["paused_or_retiring"], len(self.server.workers) - len(self.server.running)
52         )
53         yield worker_states
54 
55         if self.server.monitor.monitor_gil_contention:
56             yield CounterMetricFamily(
57                 self.build_name("gil_contention"),
58                 "GIL contention metric",
59                 value=self.server.monitor._cumulative_gil_contention,
60             )
61 
62         tasks = GaugeMetricFamily(
63             self.build_name("tasks"),
64             "Number of tasks known by scheduler",
65             labels=["state"],
66         )
67 
68         task_counter = toolz.merge_with(
69             sum, (tp.states for tp in self.server.task_prefixes.values())
70         )
71 
72         suspicious_tasks = CounterMetricFamily(
73             self.build_name("tasks_suspicious"),
74             "Total number of times a task has been marked suspicious",
75             labels=["task_prefix_name"],
76         )
77 
78         for tp in self.server.task_prefixes.values():
79             suspicious_tasks.add_metric([tp.name], tp.suspicious)
80         yield suspicious_tasks
81 
82         yield CounterMetricFamily(
83             self.build_name("tasks_forgotten"),
84             (
85                 "Total number of processed tasks no longer in memory and already "
86                 "removed from the scheduler job queue\n"
87                 "Note: Task groups on the scheduler which have all tasks "
88                 "in the forgotten state are not included."
89             ),
90             value=task_counter.get("forgotten", 0.0),
91         )
92 
93         for state in ALL_TASK_STATES:
94             if state != "forgotten":
95                 tasks.add_metric([state], task_counter.get(state, 0.0))
96         yield tasks
97 
98         time_spent_compute_tasks = CounterMetricFamily(
99             self.build_name("tasks_compute"),
100             "Total amount of compute time spent in each prefix",
101             labels=["task_prefix_name"],
102             unit="seconds",
103         )
104 
105         for tp in self.server.task_prefixes.values():
106             time_spent_compute_tasks.add_metric([tp.name], tp.all_durations["compute"])
107         yield time_spent_compute_tasks
108 
109         time_spent_transfer_tasks = CounterMetricFamily(
110             self.build_name("tasks_transfer"),
111             "Total amount of transfer time spent in each prefix",
112             labels=["task_prefix_name"],
113             unit="seconds",
114         )
115 
116         for tp in self.server.task_prefixes.values():
117             time_spent_transfer_tasks.add_metric(
118                 [tp.name], tp.all_durations["transfer"]
119             )
120         yield time_spent_transfer_tasks
121 
122         nbytes_tasks = GaugeMetricFamily(
123             self.build_name("tasks_output"),
124             "Current number of bytes in memory (without duplicates) for each prefix",
125             labels=["task_prefix_name"],
126             unit="bytes",
127         )
128         for tp in self.server.task_prefixes.values():
129             nbytes_tasks.add_metric([tp.name], tp.nbytes_total)
130         yield nbytes_tasks
131 
132         prefix_state_counts = CounterMetricFamily(
133             self.build_name("prefix_state_totals"),
134             "Accumulated count of task prefix in each state",
135             labels=["task_prefix_name", "state"],
136         )
137 
138         for tp in self.server.task_prefixes.values():
139             for state, count in tp.state_counts.items():
140                 prefix_state_counts.add_metric([tp.name, state], count)
141         yield prefix_state_counts
142 
143         now = time()
144         max_tick_duration = max(
145             self.server.digests_max["tick_duration"],
146             now - self.server._last_tick,
147         )
148         yield GaugeMetricFamily(
149             self.build_name("tick_duration_maximum_seconds"),
150             "Maximum tick duration observed since Prometheus last scraped metrics",
151             value=max_tick_duration,
152         )
153 
154         yield CounterMetricFamily(
155             self.build_name("tick_count_total"),
156             "Total number of ticks observed since the server started",
157             value=self.server._tick_counter,
158         )
159 
160         self.server.digests_max.clear()
161 
162 
163 COLLECTORS = [
164     SchedulerMetricCollector,
165     SemaphoreMetricCollector,
166     WorkStealingMetricCollector,
167 ]
168 
169 
170 class PrometheusHandler(RequestHandler):
171     _collectors = None
172 
173     def __init__(self, *args, dask_server=None, **kwargs):
174         super().__init__(*args, dask_server=dask_server, **kwargs)
175 
176         if PrometheusHandler._collectors:
177             # Especially during testing, multiple schedulers are started
178             # sequentially in the same python process
179             for _collector in PrometheusHandler._collectors:
180                 _collector.server = self.server
181             return
182 
183         PrometheusHandler._collectors = tuple(
184             collector(self.server) for collector in COLLECTORS
185         )
186         # Register collectors
187         for instantiated_collector in PrometheusHandler._collectors:
188             prometheus_client.REGISTRY.register(instantiated_collector)
189 
190     def get(self):
191         self.write(prometheus_client.generate_latest())
192         self.set_header("Content-Type", "text/plain; version=0.0.4")
193
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/distributed/http/scheduler/prometheus/core.py b/distributed/http/scheduler/prometheus/core.py
--- a/distributed/http/scheduler/prometheus/core.py
+++ b/distributed/http/scheduler/prometheus/core.py
@@ -59,6 +59,12 @@
                 value=self.server.monitor._cumulative_gil_contention,
             )
 
+        yield CounterMetricFamily(
+            self.build_name("last_time"),
+            "SystemMonitor last time",
+            value=self.server.monitor.last_time,
+        )
+
         tasks = GaugeMetricFamily(
             self.build_name("tasks"),
             "Number of tasks known by scheduler",

{"golden_diff": "diff --git a/distributed/http/scheduler/prometheus/core.py b/distributed/http/scheduler/prometheus/core.py\n--- a/distributed/http/scheduler/prometheus/core.py\n+++ b/distributed/http/scheduler/prometheus/core.py\n@@ -59,6 +59,12 @@\n                 value=self.server.monitor._cumulative_gil_contention,\n             )\n \n+        yield CounterMetricFamily(\n+            self.build_name(\"last_time\"),\n+            \"SystemMonitor last time\",\n+            value=self.server.monitor.last_time,\n+        )\n+\n         tasks = GaugeMetricFamily(\n             self.build_name(\"tasks\"),\n             \"Number of tasks known by scheduler\",\n", "issue": "Add Prometheus counter for `SystemMonitor.last_time` to improve GIL contention metric\nCurrently, the loose coupling between the system monitor's update interval and the Prometheus scraping interval can cause artifacts like a relative GIL contention > 1 (https://github.com/dask/distributed/pull/7651#issuecomment-1490571845). By exposing the system monitor's update timestamp as a Counter, we would have a synchronized timestamp available in Prometheus to serve as the basis for rate calculations. This should make such artifacts impossible.\r\n\r\ncc @ntabris, @gjoseph92, @milesgranger: Thoughts?\n", "before_files": [{"content": "from __future__ import annotations\n\nfrom collections.abc import Iterator\nfrom time import time\n\nimport prometheus_client\nimport toolz\nfrom prometheus_client.core import CounterMetricFamily, GaugeMetricFamily\n\nfrom distributed.http.prometheus import PrometheusCollector\nfrom distributed.http.scheduler.prometheus.semaphore import SemaphoreMetricCollector\nfrom distributed.http.scheduler.prometheus.stealing import WorkStealingMetricCollector\nfrom distributed.http.utils import RequestHandler\nfrom distributed.scheduler import ALL_TASK_STATES, Scheduler\n\n\nclass SchedulerMetricCollector(PrometheusCollector):\n    server: Scheduler\n\n    def __init__(self, server: Scheduler):\n        super().__init__(server)\n        self.subsystem = \"scheduler\"\n\n    def collect(self) -> Iterator[GaugeMetricFamily | CounterMetricFamily]:\n        yield GaugeMetricFamily(\n            self.build_name(\"clients\"),\n            \"Number of clients connected\",\n            value=len([k for k in self.server.clients if k != \"fire-and-forget\"]),\n        )\n\n        yield GaugeMetricFamily(\n            self.build_name(\"desired_workers\"),\n            \"Number of workers scheduler needs for task graph\",\n            value=self.server.adaptive_target(),\n        )\n\n        worker_states = GaugeMetricFamily(\n            self.build_name(\"workers\"),\n            \"Number of workers known by scheduler\",\n            labels=[\"state\"],\n        )\n        worker_states.add_metric([\"idle\"], len(self.server.idle))\n        worker_states.add_metric(\n            [\"partially_saturated\"],\n            len(self.server.running)\n            - len(self.server.idle)\n            - len(self.server.saturated),\n        )\n        worker_states.add_metric([\"saturated\"], len(self.server.saturated))\n        worker_states.add_metric(\n            [\"paused_or_retiring\"], len(self.server.workers) - len(self.server.running)\n        )\n        yield worker_states\n\n        if self.server.monitor.monitor_gil_contention:\n            yield CounterMetricFamily(\n                self.build_name(\"gil_contention\"),\n                \"GIL contention metric\",\n                value=self.server.monitor._cumulative_gil_contention,\n            )\n\n        tasks = GaugeMetricFamily(\n            self.build_name(\"tasks\"),\n            \"Number of tasks known by scheduler\",\n            labels=[\"state\"],\n        )\n\n        task_counter = toolz.merge_with(\n            sum, (tp.states for tp in self.server.task_prefixes.values())\n        )\n\n        suspicious_tasks = CounterMetricFamily(\n            self.build_name(\"tasks_suspicious\"),\n            \"Total number of times a task has been marked suspicious\",\n            labels=[\"task_prefix_name\"],\n        )\n\n        for tp in self.server.task_prefixes.values():\n            suspicious_tasks.add_metric([tp.name], tp.suspicious)\n        yield suspicious_tasks\n\n        yield CounterMetricFamily(\n            self.build_name(\"tasks_forgotten\"),\n            (\n                \"Total number of processed tasks no longer in memory and already \"\n                \"removed from the scheduler job queue\\n\"\n                \"Note: Task groups on the scheduler which have all tasks \"\n                \"in the forgotten state are not included.\"\n            ),\n            value=task_counter.get(\"forgotten\", 0.0),\n        )\n\n        for state in ALL_TASK_STATES:\n            if state != \"forgotten\":\n                tasks.add_metric([state], task_counter.get(state, 0.0))\n        yield tasks\n\n        time_spent_compute_tasks = CounterMetricFamily(\n            self.build_name(\"tasks_compute\"),\n            \"Total amount of compute time spent in each prefix\",\n            labels=[\"task_prefix_name\"],\n            unit=\"seconds\",\n        )\n\n        for tp in self.server.task_prefixes.values():\n            time_spent_compute_tasks.add_metric([tp.name], tp.all_durations[\"compute\"])\n        yield time_spent_compute_tasks\n\n        time_spent_transfer_tasks = CounterMetricFamily(\n            self.build_name(\"tasks_transfer\"),\n            \"Total amount of transfer time spent in each prefix\",\n            labels=[\"task_prefix_name\"],\n            unit=\"seconds\",\n        )\n\n        for tp in self.server.task_prefixes.values():\n            time_spent_transfer_tasks.add_metric(\n                [tp.name], tp.all_durations[\"transfer\"]\n            )\n        yield time_spent_transfer_tasks\n\n        nbytes_tasks = GaugeMetricFamily(\n            self.build_name(\"tasks_output\"),\n            \"Current number of bytes in memory (without duplicates) for each prefix\",\n            labels=[\"task_prefix_name\"],\n            unit=\"bytes\",\n        )\n        for tp in self.server.task_prefixes.values():\n            nbytes_tasks.add_metric([tp.name], tp.nbytes_total)\n        yield nbytes_tasks\n\n        prefix_state_counts = CounterMetricFamily(\n            self.build_name(\"prefix_state_totals\"),\n            \"Accumulated count of task prefix in each state\",\n            labels=[\"task_prefix_name\", \"state\"],\n        )\n\n        for tp in self.server.task_prefixes.values():\n            for state, count in tp.state_counts.items():\n                prefix_state_counts.add_metric([tp.name, state], count)\n        yield prefix_state_counts\n\n        now = time()\n        max_tick_duration = max(\n            self.server.digests_max[\"tick_duration\"],\n            now - self.server._last_tick,\n        )\n        yield GaugeMetricFamily(\n            self.build_name(\"tick_duration_maximum_seconds\"),\n            \"Maximum tick duration observed since Prometheus last scraped metrics\",\n            value=max_tick_duration,\n        )\n\n        yield CounterMetricFamily(\n            self.build_name(\"tick_count_total\"),\n            \"Total number of ticks observed since the server started\",\n            value=self.server._tick_counter,\n        )\n\n        self.server.digests_max.clear()\n\n\nCOLLECTORS = [\n    SchedulerMetricCollector,\n    SemaphoreMetricCollector,\n    WorkStealingMetricCollector,\n]\n\n\nclass PrometheusHandler(RequestHandler):\n    _collectors = None\n\n    def __init__(self, *args, dask_server=None, **kwargs):\n        super().__init__(*args, dask_server=dask_server, **kwargs)\n\n        if PrometheusHandler._collectors:\n            # Especially during testing, multiple schedulers are started\n            # sequentially in the same python process\n            for _collector in PrometheusHandler._collectors:\n                _collector.server = self.server\n            return\n\n        PrometheusHandler._collectors = tuple(\n            collector(self.server) for collector in COLLECTORS\n        )\n        # Register collectors\n        for instantiated_collector in PrometheusHandler._collectors:\n            prometheus_client.REGISTRY.register(instantiated_collector)\n\n    def get(self):\n        self.write(prometheus_client.generate_latest())\n        self.set_header(\"Content-Type\", \"text/plain; version=0.0.4\")\n", "path": "distributed/http/scheduler/prometheus/core.py"}], "after_files": [{"content": "from __future__ import annotations\n\nfrom collections.abc import Iterator\nfrom time import time\n\nimport prometheus_client\nimport toolz\nfrom prometheus_client.core import CounterMetricFamily, GaugeMetricFamily\n\nfrom distributed.http.prometheus import PrometheusCollector\nfrom distributed.http.scheduler.prometheus.semaphore import SemaphoreMetricCollector\nfrom distributed.http.scheduler.prometheus.stealing import WorkStealingMetricCollector\nfrom distributed.http.utils import RequestHandler\nfrom distributed.scheduler import ALL_TASK_STATES, Scheduler\n\n\nclass SchedulerMetricCollector(PrometheusCollector):\n    server: Scheduler\n\n    def __init__(self, server: Scheduler):\n        super().__init__(server)\n        self.subsystem = \"scheduler\"\n\n    def collect(self) -> Iterator[GaugeMetricFamily | CounterMetricFamily]:\n        yield GaugeMetricFamily(\n            self.build_name(\"clients\"),\n            \"Number of clients connected\",\n            value=len([k for k in self.server.clients if k != \"fire-and-forget\"]),\n        )\n\n        yield GaugeMetricFamily(\n            self.build_name(\"desired_workers\"),\n            \"Number of workers scheduler needs for task graph\",\n            value=self.server.adaptive_target(),\n        )\n\n        worker_states = GaugeMetricFamily(\n            self.build_name(\"workers\"),\n            \"Number of workers known by scheduler\",\n            labels=[\"state\"],\n        )\n        worker_states.add_metric([\"idle\"], len(self.server.idle))\n        worker_states.add_metric(\n            [\"partially_saturated\"],\n            len(self.server.running)\n            - len(self.server.idle)\n            - len(self.server.saturated),\n        )\n        worker_states.add_metric([\"saturated\"], len(self.server.saturated))\n        worker_states.add_metric(\n            [\"paused_or_retiring\"], len(self.server.workers) - len(self.server.running)\n        )\n        yield worker_states\n\n        if self.server.monitor.monitor_gil_contention:\n            yield CounterMetricFamily(\n                self.build_name(\"gil_contention\"),\n                \"GIL contention metric\",\n                value=self.server.monitor._cumulative_gil_contention,\n            )\n\n        yield CounterMetricFamily(\n            self.build_name(\"last_time\"),\n            \"SystemMonitor last time\",\n            value=self.server.monitor.last_time,\n        )\n\n        tasks = GaugeMetricFamily(\n            self.build_name(\"tasks\"),\n            \"Number of tasks known by scheduler\",\n            labels=[\"state\"],\n        )\n\n        task_counter = toolz.merge_with(\n            sum, (tp.states for tp in self.server.task_prefixes.values())\n        )\n\n        suspicious_tasks = CounterMetricFamily(\n            self.build_name(\"tasks_suspicious\"),\n            \"Total number of times a task has been marked suspicious\",\n            labels=[\"task_prefix_name\"],\n        )\n\n        for tp in self.server.task_prefixes.values():\n            suspicious_tasks.add_metric([tp.name], tp.suspicious)\n        yield suspicious_tasks\n\n        yield CounterMetricFamily(\n            self.build_name(\"tasks_forgotten\"),\n            (\n                \"Total number of processed tasks no longer in memory and already \"\n                \"removed from the scheduler job queue\\n\"\n                \"Note: Task groups on the scheduler which have all tasks \"\n                \"in the forgotten state are not included.\"\n            ),\n            value=task_counter.get(\"forgotten\", 0.0),\n        )\n\n        for state in ALL_TASK_STATES:\n            if state != \"forgotten\":\n                tasks.add_metric([state], task_counter.get(state, 0.0))\n        yield tasks\n\n        time_spent_compute_tasks = CounterMetricFamily(\n            self.build_name(\"tasks_compute\"),\n            \"Total amount of compute time spent in each prefix\",\n            labels=[\"task_prefix_name\"],\n            unit=\"seconds\",\n        )\n\n        for tp in self.server.task_prefixes.values():\n            time_spent_compute_tasks.add_metric([tp.name], tp.all_durations[\"compute\"])\n        yield time_spent_compute_tasks\n\n        time_spent_transfer_tasks = CounterMetricFamily(\n            self.build_name(\"tasks_transfer\"),\n            \"Total amount of transfer time spent in each prefix\",\n            labels=[\"task_prefix_name\"],\n            unit=\"seconds\",\n        )\n\n        for tp in self.server.task_prefixes.values():\n            time_spent_transfer_tasks.add_metric(\n                [tp.name], tp.all_durations[\"transfer\"]\n            )\n        yield time_spent_transfer_tasks\n\n        nbytes_tasks = GaugeMetricFamily(\n            self.build_name(\"tasks_output\"),\n            \"Current number of bytes in memory (without duplicates) for each prefix\",\n            labels=[\"task_prefix_name\"],\n            unit=\"bytes\",\n        )\n        for tp in self.server.task_prefixes.values():\n            nbytes_tasks.add_metric([tp.name], tp.nbytes_total)\n        yield nbytes_tasks\n\n        prefix_state_counts = CounterMetricFamily(\n            self.build_name(\"prefix_state_totals\"),\n            \"Accumulated count of task prefix in each state\",\n            labels=[\"task_prefix_name\", \"state\"],\n        )\n\n        for tp in self.server.task_prefixes.values():\n            for state, count in tp.state_counts.items():\n                prefix_state_counts.add_metric([tp.name, state], count)\n        yield prefix_state_counts\n\n        now = time()\n        max_tick_duration = max(\n            self.server.digests_max[\"tick_duration\"],\n            now - self.server._last_tick,\n        )\n        yield GaugeMetricFamily(\n            self.build_name(\"tick_duration_maximum_seconds\"),\n            \"Maximum tick duration observed since Prometheus last scraped metrics\",\n            value=max_tick_duration,\n        )\n\n        yield CounterMetricFamily(\n            self.build_name(\"tick_count_total\"),\n            \"Total number of ticks observed since the server started\",\n            value=self.server._tick_counter,\n        )\n\n        self.server.digests_max.clear()\n\n\nCOLLECTORS = [\n    SchedulerMetricCollector,\n    SemaphoreMetricCollector,\n    WorkStealingMetricCollector,\n]\n\n\nclass PrometheusHandler(RequestHandler):\n    _collectors = None\n\n    def __init__(self, *args, dask_server=None, **kwargs):\n        super().__init__(*args, dask_server=dask_server, **kwargs)\n\n        if PrometheusHandler._collectors:\n            # Especially during testing, multiple schedulers are started\n            # sequentially in the same python process\n            for _collector in PrometheusHandler._collectors:\n                _collector.server = self.server\n            return\n\n        PrometheusHandler._collectors = tuple(\n            collector(self.server) for collector in COLLECTORS\n        )\n        # Register collectors\n        for instantiated_collector in PrometheusHandler._collectors:\n            prometheus_client.REGISTRY.register(instantiated_collector)\n\n    def get(self):\n        self.write(prometheus_client.generate_latest())\n        self.set_header(\"Content-Type\", \"text/plain; version=0.0.4\")\n", "path": "distributed/http/scheduler/prometheus/core.py"}]}