Add transfer_outgoing_bytes_total metric

Closes dask#7123

distributed/http/worker/prometheus/core.py

@@ -112,6 +112,15 @@ def collect(self):
             value=self.server.transfer_outgoing_count,
         )
 
+        yield CounterMetricFamily(
+            self.build_name("transfer_outgoing_bytes_total"),
+            (
+                "Total size of data transferred to other workers "
+                "since the worker was started"
+            ),
+            value=self.server.transfer_outgoing_bytes_total,
+        )
+
         yield CounterMetricFamily(
             self.build_name("transfer_outgoing_count_total"),
             (

distributed/http/worker/tests/test_worker_http.py

@@ -36,6 +36,7 @@ async def test_prometheus(c, s, a):
         "dask_worker_transfer_outgoing_bytes",
         "dask_worker_transfer_outgoing_count",
         "dask_worker_transfer_outgoing_count_total",
+        "dask_worker_transfer_outgoing_bytes_total",
         "dask_worker_tick_count_total",
         "dask_worker_tick_duration_maximum_seconds",
     }

distributed/worker.py

@@ -394,6 +394,8 @@ class Worker(BaseWorker, ServerNode):
     transfer_outgoing_log: deque[dict[str, Any]]
     #: Total number of data transfers to other workers since the worker was started
     transfer_outgoing_count_total: int
+    #: Total size of data transferred to other workers (including in-progress transfers)
+    transfer_outgoing_bytes_total: int
     #: Current total size of open data transfers to other workers
     transfer_outgoing_bytes: int
     #: Current number of open data transfers to other workers
@@ -556,6 +558,7 @@ def __init__(
         self.transfer_incoming_log = deque(maxlen=100000)
         self.transfer_outgoing_log = deque(maxlen=100000)
         self.transfer_outgoing_count_total = 0
+        self.transfer_outgoing_bytes_total = 0
         self.transfer_outgoing_bytes = 0
         self.transfer_outgoing_count = 0
         self.bandwidth = parse_bytes(dask.config.get("distributed.scheduler.bandwidth"))
@@ -1746,6 +1749,7 @@ async def get_data(
         bytes_per_task = {k: self.state.tasks[k].nbytes or 0 for k in data}
         total_bytes = sum(bytes_per_task.values())
         self.transfer_outgoing_bytes += total_bytes
+        self.transfer_outgoing_bytes_total += total_bytes
         stop = time()
         if self.digests is not None:
             self.digests["get-data-load-duration"].add(stop - start)