diff --git a/docs/metrics.md b/docs/metrics.md index 05559dd5..327f6079 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -60,6 +60,8 @@ Steps: |---------------------|---------|-------------| | `TQ_METRICS_COLLECT_INTERVAL` | `10` | Background collection interval (seconds) | | `TQ_METRICS_STORAGE_TIMEOUT` | `5` | ZMQ timeout for storage unit queries (seconds) | +| `TQ_ACCEPT_PROBE_INTERVAL` | `0` (off) | Accept-queue sampling period (seconds); enables the accept-queue metrics below | +| `TQ_STORAGE_ZMQ_BACKLOG` | `4096` | Accept-queue depth for the storage unit's listening socket | ## Architecture @@ -140,6 +142,38 @@ Steps: | `tq_storage_request_latency_p50` | Gauge | `storage_unit_id`, `op_type` | P50 request latency (seconds) | | `tq_storage_request_latency_p99` | Gauge | `storage_unit_id`, `op_type` | P99 request latency (seconds) | +### Storage Request-Loss Diagnostics (collected via ZMQ, exposed on controller) + +Arrivals are counted when the worker decodes a request, while `tq_storage_request_ops` only +advances once one completes, so a sustained gap between them means requests are arriving and +not finishing. The counters are cumulative per operation and carry no request identity, so +they characterise a unit, not any individual request. + +| Metric | Type | Labels | Description | +|--------|------|--------|-------------| +| `tq_storage_requests_arrived` | Gauge | `storage_unit_id` | Requests decoded by the worker, whether or not they completed | +| `tq_storage_arrivals_by_op` | Gauge | `storage_unit_id`, `op_type` | Same, broken down by operation | + +The accept-queue series below exist only when the unit runs with `TQ_ACCEPT_PROBE_INTERVAL` +set; the series are removed rather than reported as zero when the probe is off, so a missing +series means "not measured" rather than "no drops". `tq_storage_accept_queue_peak` and the two +drop counters are cumulative since the probe started, so use `rate()` on them. + +| Metric | Type | Labels | Description | +|--------|------|--------|-------------| +| `tq_storage_accept_queue_backlog` | Gauge | `storage_unit_id` | Configured accept-queue depth (`TQ_STORAGE_ZMQ_BACKLOG`) | +| `tq_storage_accept_queue_peak` | Gauge | `storage_unit_id` | Deepest accept-queue occupancy seen by the probe | +| `tq_storage_accept_queue_peak_utilization_ratio` | Gauge | `storage_unit_id` | Peak occupancy as a fraction of the backlog | +| `tq_storage_socket_drops` | Gauge | `storage_unit_id` | Connections dropped on this listening socket since the probe started | +| `tq_storage_listen_overflows` | Gauge | `storage_unit_id` | Namespace-wide accept-queue overflows since the probe started | +| `tq_storage_listen_other_drops` | Gauge | `storage_unit_id` | Namespace-wide establishment drops that were **not** overflows | + +The kernel charges a listening socket's `sk_drops` on several connection-establishment +failures — a full accept queue, but also failures to allocate or route the new connection — so +`tq_storage_socket_drops` rising locates the socket, not the cause. Compare the last two +series: a non-zero `tq_storage_listen_other_drops` means raising the backlog would not have +prevented every drop in that window. + ### Storage Unit Native Metrics (exposed on each storage unit's own endpoint) | Metric | Type | Labels | Description | @@ -163,6 +197,7 @@ The dashboard ([`scripts/grafana_dashboard.json`](../scripts/grafana_dashboard.j | **Request Throughput & Latency** | Controller Request Rate (ops/s), Controller Request Latency (repeats per quantile) | | **Partition Status** | Samples per Partition, Production Progress, Consumption Progress | | **Storage Units** | Utilization Bar Gauge, Active Keys, Capacity vs Active Keys, RSS Memory, Storage Request Rate, Storage Request Latency (repeats per quantile), Produced vs Cleared Samples/s, Active Keys Delta | +| **Storage Request-Loss Diagnostics** | Arrived vs Completed/s, Accept-Queue Peak vs Backlog, Listening-Socket Drops/s, Overflow vs Other Establishment Drops/s | ### Template Variables diff --git a/scripts/grafana_dashboard.json b/scripts/grafana_dashboard.json index 0e6c5341..9f3ee36b 100644 --- a/scripts/grafana_dashboard.json +++ b/scripts/grafana_dashboard.json @@ -370,6 +370,68 @@ { "expr": "sum(tq_storage_active_keys_total)", "legendFormat": "Total Active Keys (all storage units)" } ], "type": "timeseries" + }, + { + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 86 }, + "id": 104, + "title": "Storage Request-Loss Diagnostics", + "type": "row" + }, + { + "datasource": { "type": "prometheus", "uid": "${datasource}" }, + "description": "Arrivals are counted when the worker decodes a request; completions only when it finishes. A sustained gap means requests are arriving and not completing. Neither series identifies an individual request.", + "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, "custom": { "fillOpacity": 10, "lineWidth": 2 } } }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 87 }, + "id": 40, + "options": { "legend": { "calcs": ["lastNotNull"], "displayMode": "table", "placement": "bottom" }, "tooltip": { "mode": "multi" } }, + "title": "Arrived vs Completed (per second)", + "targets": [ + { "expr": "sum by (op_type) (rate(tq_storage_arrivals_by_op{op_type=~\"$op_type\"}[$__rate_interval]))", "legendFormat": "arrived {{ op_type }}" }, + { "expr": "sum by (op_type) (rate(tq_storage_request_ops{op_type=~\"$op_type\"}[$__rate_interval]))", "legendFormat": "completed {{ op_type }}" } + ], + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${datasource}" }, + "description": "Peak is a high-water mark since the probe started, so it never decreases. Only present when TQ_ACCEPT_PROBE_INTERVAL is set.", + "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, "custom": { "fillOpacity": 10, "lineWidth": 2 } } }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 87 }, + "id": 41, + "options": { "legend": { "calcs": ["lastNotNull"], "displayMode": "table", "placement": "bottom" }, "tooltip": { "mode": "multi" } }, + "title": "Accept-Queue Peak vs Backlog", + "targets": [ + { "expr": "tq_storage_accept_queue_peak", "legendFormat": "peak {{ storage_unit_id }}" }, + { "expr": "tq_storage_accept_queue_backlog", "legendFormat": "backlog {{ storage_unit_id }}" } + ], + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${datasource}" }, + "description": "The kernel charges sk_drops for several connection-establishment failures, of which a full accept queue is only one, so a rise locates the socket, not the cause.", + "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, "custom": { "fillOpacity": 10, "lineWidth": 2 } } }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 95 }, + "id": 42, + "options": { "legend": { "calcs": ["lastNotNull"], "displayMode": "table", "placement": "bottom" }, "tooltip": { "mode": "multi" } }, + "title": "Listening-Socket Drops (per second)", + "targets": [ + { "expr": "sum by (storage_unit_id) (rate(tq_storage_socket_drops[$__rate_interval]))", "legendFormat": "{{ storage_unit_id }}" } + ], + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "${datasource}" }, + "description": "Namespace-wide, not per port. A non-zero 'other' series means raising the backlog would not have prevented every drop in the window.", + "fieldConfig": { "defaults": { "color": { "mode": "palette-classic" }, "custom": { "fillOpacity": 10, "lineWidth": 2 } } }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 95 }, + "id": 43, + "options": { "legend": { "calcs": ["lastNotNull"], "displayMode": "table", "placement": "bottom" }, "tooltip": { "mode": "multi" } }, + "title": "Overflow vs Other Establishment Drops (per second)", + "targets": [ + { "expr": "sum (rate(tq_storage_listen_overflows[$__rate_interval]))", "legendFormat": "accept-queue overflow" }, + { "expr": "sum (rate(tq_storage_listen_other_drops[$__rate_interval]))", "legendFormat": "other establishment drops" } + ], + "type": "timeseries" } ], "refresh": "10s", diff --git a/tests/test_accept_probe.py b/tests/test_accept_probe.py new file mode 100644 index 00000000..a834955c --- /dev/null +++ b/tests/test_accept_probe.py @@ -0,0 +1,257 @@ +# Copyright 2025 Huawei Technologies Co., Ltd. All Rights Reserved. +# Copyright 2025 The TransferQueue Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for the accept-queue probe. + +A connection dropped during establishment is invisible to both ends, and the kernel +charges the listening socket's sk_drops on several such paths, of which a full accept +queue is only one. These tests pin the arithmetic the probe reports, the levels it logs +at, and that it never names a cause its counters cannot establish. +""" + +import threading + +from transfer_queue.utils.accept_probe import ( + AcceptQueueProbe, + AcceptQueueSample, + AcceptQueueStats, + _read_listen_overflows, + sample_accept_queue, +) + + +def _sample( + recv_q: int, backlog: int = 100, sk_drops: int = 0, overflows: int = 0, drops: int | None = None +) -> AcceptQueueSample: + return AcceptQueueSample( + timestamp=0.0, + recv_q=recv_q, + backlog=backlog, + sk_drops=sk_drops, + listen_overflows=overflows, + listen_drops=overflows if drops is None else drops, + ) + + +def _probe() -> AcceptQueueProbe: + return AcceptQueueProbe(port=34513, owner_id="TQ_STORAGE_UNIT_test", interval_s=0.01) + + +def test_utilization_is_depth_over_backlog(): + assert _sample(50, backlog=100).utilization == 0.5 + + +def test_zero_backlog_reports_zero_utilization_not_zero_division(): + """A socket read before bind reports backlog 0; that must not raise.""" + assert _sample(5, backlog=0).utilization == 0.0 + + +def test_full_queue_reports_full_utilization(): + assert _sample(100, backlog=100).utilization == 1.0 + + +def test_empty_stats_report_zero_deltas(): + stats = AcceptQueueStats(port=1234) + + assert stats.sk_drops_delta == 0 + assert stats.overflow_delta == 0 + + +def test_drop_delta_is_last_minus_first(): + """Deltas are what matter: the absolute counters carry the whole uptime's history.""" + stats = AcceptQueueStats(port=1234) + stats.first_sample = _sample(0, sk_drops=23, overflows=1775) + stats.last_sample = _sample(0, sk_drops=31, overflows=1790) + + assert stats.sk_drops_delta == 8 + assert stats.overflow_delta == 15 + + +def test_describe_names_port_and_deltas(): + stats = AcceptQueueStats(port=34513, backlog=100, peak_recv_q=97) + stats.first_sample = _sample(0, sk_drops=23) + stats.last_sample = _sample(0, sk_drops=24) + + text = stats.describe() + + assert "port=34513" in text + assert "peak_recv_q=97" in text + assert "sk_drops_delta=1" in text + + +def test_peak_tracks_highest_depth_not_latest(): + """The burst is the signal; sampling after it drains reads zero.""" + probe = _probe() + + probe._record(_sample(10)) + probe._record(_sample(97)) + probe._record(_sample(3)) + + assert probe.stats.peak_recv_q == 97 + + +def test_first_sample_is_retained_as_delta_baseline(): + probe = _probe() + + probe._record(_sample(1, sk_drops=23)) + probe._record(_sample(2, sk_drops=25)) + + assert probe.stats.first_sample.sk_drops == 23 + assert probe.stats.sk_drops_delta == 2 + + +def test_drop_increase_is_logged_at_error(caplog): + """A drop is the direct evidence, so it must not be buried at debug level.""" + probe = _probe() + probe._record(_sample(0, sk_drops=23)) + + with caplog.at_level("ERROR"): + probe._record(_sample(100, sk_drops=24)) + + assert "dropped an incoming connection" in caplog.text + + +def test_steady_drop_count_is_logged_once_not_every_sample(caplog): + """sk_drops is cumulative, so a past drop must not be re-reported forever. + + Measuring against the probe's first sample made the condition permanently true once + the socket had ever dropped a connection: at a 0.1s interval that is ten errors a + second for the life of the process, and it erases when the drop actually happened. + """ + probe = _probe() + probe._record(_sample(0, sk_drops=12)) + + with caplog.at_level("ERROR"): + probe._record(_sample(0, sk_drops=13)) # a new drop -- report it + for _ in range(20): + probe._record(_sample(0, sk_drops=13)) # unchanged -- stay quiet + + assert caplog.text.count("dropped an incoming connection") == 1 + + +def test_each_new_drop_is_reported(caplog): + """Quieting the repeat must not swallow genuinely new drops.""" + probe = _probe() + probe._record(_sample(0, sk_drops=12)) + + with caplog.at_level("ERROR"): + probe._record(_sample(0, sk_drops=13)) + probe._record(_sample(0, sk_drops=13)) + probe._record(_sample(0, sk_drops=14)) + + assert caplog.text.count("dropped an incoming connection") == 2 + + +def test_near_full_queue_warns_once(caplog): + """Repeating the warning every 100ms would flood the log of a 512-node job.""" + probe = _probe() + + with caplog.at_level("WARNING"): + probe._record(_sample(60)) + probe._record(_sample(70)) + + assert caplog.text.count("reached") == 1 + + +def test_no_warning_below_threshold(): + probe = _probe() + + probe._record(_sample(10)) + + assert probe._warned is False + + +def test_listen_overflows_returns_two_non_negative_ints(): + """Reads the real host, so assert shape rather than a specific value.""" + overflows, drops = _read_listen_overflows() + + assert overflows >= 0 + assert drops >= 0 + + +def test_sampling_an_unused_port_returns_none(): + assert sample_accept_queue(1) is None + + +def test_unit_shutdown_stops_the_probe(): + """Nothing else calls stop(), so the sampling thread would outlive the unit.""" + from unittest.mock import MagicMock + + from transfer_queue.storage.simple_storage import SimpleStorageUnit + + unit_class = SimpleStorageUnit.__ray_metadata__.modified_class + probe = MagicMock() + + unit_class._shutdown_resources( + shutdown_event=threading.Event(), + worker_thread=None, + proxy_thread=None, + zmq_context=None, + put_get_socket=None, + accept_probe=probe, + ) + + probe.stop.assert_called_once() + + +def test_shutdown_without_a_probe_is_a_no_op(): + """The probe is opt-in, so the default path must not require one.""" + from transfer_queue.storage.simple_storage import SimpleStorageUnit + + unit_class = SimpleStorageUnit.__ray_metadata__.modified_class + + unit_class._shutdown_resources( + shutdown_event=threading.Event(), + worker_thread=None, + proxy_thread=None, + zmq_context=None, + put_get_socket=None, + ) + + +def test_non_overflow_drops_are_separated_from_overflows(): + """ListenDrops counts every establishment failure; only some are queue overflows. + + The kernel charges a listening socket's sk_drops on several paths -- a full accept + queue, but also failures to allocate or route the new connection -- so this difference + is what says whether a bigger backlog could have helped. + """ + stats = AcceptQueueStats(port=1234) + stats.first_sample = _sample(0, overflows=10, drops=20) + stats.last_sample = _sample(0, overflows=11, drops=25) + + assert stats.overflow_delta == 1 + assert stats.non_overflow_drop_delta == 4 + + +def test_pure_overflow_window_reports_no_other_drops(): + stats = AcceptQueueStats(port=1234) + stats.first_sample = _sample(0, overflows=10, drops=10) + stats.last_sample = _sample(0, overflows=13, drops=13) + + assert stats.overflow_delta == 3 + assert stats.non_overflow_drop_delta == 0 + + +def test_drop_alert_does_not_assert_the_queue_overflowed(caplog): + """sk_drops locates the socket, not the cause, so the alert must not name one.""" + probe = _probe() + probe._record(_sample(0, sk_drops=1)) + + with caplog.at_level("ERROR"): + probe._record(_sample(0, sk_drops=2)) + + assert "dropped an incoming connection" in caplog.text + assert "accept queue dropped" not in caplog.text diff --git a/tests/test_metrics.py b/tests/test_metrics.py index a8ac3e2b..66e82a3f 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -98,6 +98,14 @@ def test_all_metrics_are_registered(self): "tq_storage_active_keys_total", "tq_storage_utilization_ratio", "tq_storage_memory_rss_bytes", + "tq_storage_requests_arrived", + "tq_storage_arrivals_by_op", + "tq_storage_accept_queue_backlog", + "tq_storage_accept_queue_peak", + "tq_storage_accept_queue_peak_utilization_ratio", + "tq_storage_socket_drops", + "tq_storage_listen_overflows", + "tq_storage_listen_other_drops", ] registered = {m.name for m in exporter.registry.collect()} @@ -315,6 +323,76 @@ def test_storage_metrics_populated_on_success(self): assert exporter.storage_utilization.labels(storage_unit_id="SU_001")._value.get() == 0.25 assert exporter.storage_memory_rss.labels(storage_unit_id="SU_001")._value.get() == 512 * 1024 * 1024 + def test_arrival_counters_are_exported(self): + """Arrival counts reach Prometheus, so a dashboard can compare them with completions.""" + exporter = TQMetricsExporter() + fake_su_info = MagicMock() + fake_su_info.id = "SU_001" + exporter._storage_unit_infos = {"SU_001": fake_su_info} + exporter._query_storage_unit = MagicMock( + return_value={ + "storage_unit_id": "SU_001", + "capacity": 1000, + "active_keys": 1, + "requests_arrived": 42, + "arrivals_by_op": {"GET_DATA": 30, "PUT_DATA": 12}, + } + ) + + exporter.collect_storage_metrics() + + assert exporter.storage_requests_arrived.labels(storage_unit_id="SU_001")._value.get() == 42 + by_op = exporter.storage_arrivals_by_op + assert by_op.labels(storage_unit_id="SU_001", op_type="GET_DATA")._value.get() == 30 + assert by_op.labels(storage_unit_id="SU_001", op_type="PUT_DATA")._value.get() == 12 + + def test_accept_queue_metrics_are_exported(self): + """The overflow/non-overflow split is what tells a dashboard if backlog is the issue.""" + exporter = TQMetricsExporter() + fake_su_info = MagicMock() + fake_su_info.id = "SU_001" + exporter._storage_unit_infos = {"SU_001": fake_su_info} + exporter._query_storage_unit = MagicMock( + return_value={ + "storage_unit_id": "SU_001", + "capacity": 1000, + "active_keys": 1, + "accept_queue": { + "backlog": 4096, + "peak_recv_q": 97, + "peak_utilization": 0.02, + "sk_drops_delta": 5, + "listen_overflow_delta": 2, + "non_overflow_drop_delta": 3, + }, + } + ) + + exporter.collect_storage_metrics() + + label = {"storage_unit_id": "SU_001"} + assert exporter.storage_accept_queue_backlog.labels(**label)._value.get() == 4096 + assert exporter.storage_accept_queue_peak.labels(**label)._value.get() == 97 + assert exporter.storage_socket_drops.labels(**label)._value.get() == 5 + assert exporter.storage_listen_overflows.labels(**label)._value.get() == 2 + assert exporter.storage_listen_other_drops.labels(**label)._value.get() == 3 + + def test_accept_queue_series_absent_when_the_probe_is_off(self): + """The probe is opt-in; reporting 0 drops would read as 'measured, and none'.""" + exporter = TQMetricsExporter() + fake_su_info = MagicMock() + fake_su_info.id = "SU_001" + exporter._storage_unit_infos = {"SU_001": fake_su_info} + exporter._query_storage_unit = MagicMock( + return_value={"storage_unit_id": "SU_001", "capacity": 1000, "active_keys": 1} + ) + + exporter.collect_storage_metrics() + + exported = {sample.name for metric in exporter.registry.collect() for sample in metric.samples} + assert "tq_storage_socket_drops" not in exported + assert "tq_storage_accept_queue_backlog" not in exported + def test_storage_metrics_handles_query_failure(self): """If a storage unit query fails, other units should still be collected.""" exporter = TQMetricsExporter() diff --git a/tests/test_storage_request_retry.py b/tests/test_storage_request_retry.py index ec6e9476..59f82edc 100644 --- a/tests/test_storage_request_retry.py +++ b/tests/test_storage_request_retry.py @@ -147,7 +147,11 @@ def test_log_heavy_operation_thresholds(caplog, elapsed, payload_bytes, should_l @pytest.mark.parametrize( "tcp_result, probe_result, expected", [ - ((None, None), {"active_keys": 4, "op_stats": {"GET_DATA": {"request_count": 1025}}}, "request_lost"), + # A probe that answers proves only that the unit is serving again, whatever the + # cumulative counters happen to say. + ((None, None), {"active_keys": 4, "arrivals_by_op": {"GET_DATA": 0}}, "unit_serving_again"), + ((None, None), {"active_keys": 4, "arrivals_by_op": {"GET_DATA": 9}}, "unit_serving_again"), + ((None, None), {"active_keys": 4}, "unit_serving_again"), ((None, None), zmq.error.Again(), "unit_not_serving"), (ConnectionRefusedError(), zmq.error.Again(), "tcp=down(ConnectionRefusedError)"), ], @@ -170,6 +174,41 @@ async def test_diagnosis_classifies_the_failure(tcp_result, probe_result, expect assert expected in diagnosis +def test_diagnosis_does_not_report_empty_op_stats_as_zero_traffic(): + """op_stats is Prometheus-gated, so an empty dict must not read as 'served nothing'.""" + described = ssm._describe_unit_state({"requests_arrived": 7, "active_keys": 1}) + + assert "completed=unavailable(prometheus_disabled)" in described + assert "completed={}" not in described + + +@pytest.mark.parametrize( + "arrivals", + [ + # Nine historical GETs, all already completed: they say nothing about this request. + {"GET_DATA": 9}, + # The unit restarted during the timeout window, so its counters began again at zero. + {"GET_DATA": 0}, + # The request arrived and finished after the caller gave up waiting. + {"GET_DATA": 10}, + ], +) +def test_cumulative_counters_never_become_a_claim_about_this_request(arrivals): + """Counters are per-op and cumulative, so no value of them locates one request. + + They carry no request id, never decrease, and reset when the unit restarts, so both + directions of inference are unsound: a nonzero count may be entirely historical, and a + zero one may mean the request reached a previous process. + """ + described = ssm._describe_unit_state({"requests_arrived": sum(arrivals.values()), "arrivals_by_op": arrivals}) + + assert "verdict=unit_serving_again" in described + assert "arrived_but_unfinished" not in described + assert "request_lost_in_flight" not in described + # The raw counters stay in the line as triage input, just not as a verdict. + assert "arrivals_by_op=" in described + + @pytest.mark.asyncio async def test_diagnosis_handles_an_unknown_unit(): assert "unit_not_registered" in await _manager(with_unit=False)._diagnose_storage_unit("missing") diff --git a/transfer_queue/metrics.py b/transfer_queue/metrics.py index 12914585..e75792b5 100644 --- a/transfer_queue/metrics.py +++ b/transfer_queue/metrics.py @@ -185,6 +185,62 @@ def _define_controller_metrics(self) -> None: "tq_storage_memory_rss_bytes", "Storage unit process RSS memory", ["storage_unit_id"], registry=r ) + # ---- Storage-unit request-loss diagnostics ---- + # Read against tq_storage_request_ops, which advances only on completion: a gap + # between the two is requests arriving and not finishing. + self.storage_requests_arrived = Gauge( + "tq_storage_requests_arrived", + "Requests decoded by the storage unit worker, whether or not they completed", + ["storage_unit_id"], + registry=r, + ) + self.storage_arrivals_by_op = Gauge( + "tq_storage_arrivals_by_op", + "Requests decoded by the storage unit worker, by operation", + ["storage_unit_id", "op_type"], + registry=r, + ) + + # ---- Accept-queue probe (only populated when TQ_ACCEPT_PROBE_INTERVAL > 0) ---- + self.storage_accept_queue_backlog = Gauge( + "tq_storage_accept_queue_backlog", + "Configured accept-queue depth of the storage unit's listening socket", + ["storage_unit_id"], + registry=r, + ) + self.storage_accept_queue_peak = Gauge( + "tq_storage_accept_queue_peak", + "Deepest accept-queue occupancy seen by the probe", + ["storage_unit_id"], + registry=r, + ) + self.storage_accept_queue_peak_utilization = Gauge( + "tq_storage_accept_queue_peak_utilization_ratio", + "Peak accept-queue occupancy as a fraction of the backlog", + ["storage_unit_id"], + registry=r, + ) + self.storage_socket_drops = Gauge( + "tq_storage_socket_drops", + "Connections dropped on this listening socket since the probe started", + ["storage_unit_id"], + registry=r, + ) + # Split because sk_drops covers several establishment failures, not only a full + # queue; the difference is what says whether a bigger backlog would have helped. + self.storage_listen_overflows = Gauge( + "tq_storage_listen_overflows", + "Namespace-wide accept-queue overflows since the probe started", + ["storage_unit_id"], + registry=r, + ) + self.storage_listen_other_drops = Gauge( + "tq_storage_listen_other_drops", + "Namespace-wide establishment drops that were not accept-queue overflows", + ["storage_unit_id"], + registry=r, + ) + # ---- Storage request metrics (collected via ZMQ, exposed as gauges) ---- # P50/P99 are pre-computed on the storage unit side and sent via ZMQ, # avoiding the need to replicate histogram bucket structures (which @@ -362,6 +418,30 @@ def collect_storage_metrics(self) -> None: self.storage_active_keys.labels(storage_unit_id=label).set(active) self.storage_memory_rss.labels(storage_unit_id=label).set(metrics.get("process_rss_bytes", 0)) + self.storage_requests_arrived.labels(storage_unit_id=label).set(metrics.get("requests_arrived", 0)) + for op_type, arrived in (metrics.get("arrivals_by_op") or {}).items(): + self.storage_arrivals_by_op.labels(storage_unit_id=label, op_type=op_type).set(arrived) + + # Drop the series rather than report zero when the probe is off, so a + # disabled probe is not read as "measured, and no drops". + accept_queue = metrics.get("accept_queue") + accept_gauges = ( + (self.storage_accept_queue_backlog, "backlog"), + (self.storage_accept_queue_peak, "peak_recv_q"), + (self.storage_accept_queue_peak_utilization, "peak_utilization"), + (self.storage_socket_drops, "sk_drops_delta"), + (self.storage_listen_overflows, "listen_overflow_delta"), + (self.storage_listen_other_drops, "non_overflow_drop_delta"), + ) + for gauge, key in accept_gauges: + if accept_queue is None: + try: + gauge.remove(label) + except (KeyError, ValueError): + pass + else: + gauge.labels(storage_unit_id=label).set(accept_queue.get(key, 0)) + # Per-operation request stats for op_type, op_data in metrics.get("op_stats", {}).items(): self.storage_request_ops.labels(storage_unit_id=label, op_type=op_type).set( diff --git a/transfer_queue/storage/managers/simple_storage_manager.py b/transfer_queue/storage/managers/simple_storage_manager.py index 70765cb9..c89d1cfb 100644 --- a/transfer_queue/storage/managers/simple_storage_manager.py +++ b/transfer_queue/storage/managers/simple_storage_manager.py @@ -63,11 +63,30 @@ class StorageUnitTimeout(RuntimeError): """A storage unit did not answer within the send/recv timeout. Distinct from an error the unit reported: only a missing answer is worth a new connection. - The message must name the unit, its endpoint and the timeout, because it is what the callers - of ``put_data`` and ``get_data`` see, and the retry logs rely on it instead of repeating them. """ +def _describe_unit_state(body: dict[str, Any]) -> str: + """Summarize a successful probe: the unit is serving again, plus its own counters. + + Draws no conclusion about the timed-out request: the counters are cumulative per + operation and carry no request identity, so no value of them locates one request. + """ + parts = [ + f"requests_arrived={body.get('requests_arrived')}", + f"arrivals_by_op={body.get('arrivals_by_op')}", + f"active_keys={body.get('active_keys')}", + f"rss_gb={body.get('process_rss_bytes', 0) / 2**30:.2f}", + ] + op_stats = body.get("op_stats") or {} + if op_stats: + parts.append(f"completed={ {op: stats.get('request_count') for op, stats in op_stats.items()} }") + else: + # Only with Prometheus enabled; an empty dict would read as "served nothing". + parts.append("completed=unavailable(prometheus_disabled)") + return f"verdict=unit_serving_again ({' '.join(parts)})" + + _SU_SUBDIR = "simple_storage" _SU_INFO_FILE = "storage_unit_info.json" @@ -240,9 +259,10 @@ async def _probe_storage_unit(self, target_storage_unit: str, socket: zmq.Socket return response_msg.body async def _diagnose_storage_unit(self, target_storage_unit: str) -> str: - """Classify a timeout as a lost request, a stuck unit, or an unreachable node. + """Report whether the unit is reachable and serving after a request to it timed out. Returns one log line and never raises: it runs while another failure is being reported. + Says nothing about where that request went; no per-request state exists to show it. """ info = self.storage_unit_infos.get(target_storage_unit) if info is None: @@ -259,17 +279,13 @@ async def _diagnose_storage_unit(self, target_storage_unit: str) -> str: try: body = await self._probe_storage_unit(target_storage_unit=target_storage_unit) - op_counts = {op: stats.get("request_count") for op, stats in (body.get("op_stats") or {}).items()} - return ( - f"{tcp} verdict=request_lost_in_flight (unit answered a fresh probe: " - f"ops={op_counts} active_keys={body.get('active_keys')} " - f"rss_gb={body.get('process_rss_bytes', 0) / 2**30:.2f})" - ) except zmq.error.Again: return f"{tcp} verdict=unit_not_serving (no probe answer in {TQ_SIMPLE_STORAGE_PROBE_TIMEOUT}s)" except Exception as e: return f"{tcp} verdict=unknown (probe failed: {type(e).__name__}: {e})" + return f"{tcp} {_describe_unit_state(body)}" + async def _request_with_retry( self, operation: str, diff --git a/transfer_queue/storage/simple_storage.py b/transfer_queue/storage/simple_storage.py index cd0d2b4b..61c5e9a1 100644 --- a/transfer_queue/storage/simple_storage.py +++ b/transfer_queue/storage/simple_storage.py @@ -43,6 +43,7 @@ if TYPE_CHECKING: from transfer_queue.metrics import TQMetricsExporter + from transfer_queue.utils.accept_probe import AcceptQueueProbe logger = get_logger(__name__) @@ -52,6 +53,12 @@ # Marks a GET_ERROR reply as "the key is gone" so the caller can tell it apart from a real fault. KEY_NOT_FOUND_MARKER = "TQKeyNotFound" +# Accept-queue depth for the client-facing ROUTER. A full queue loses connections silently. +TQ_STORAGE_ZMQ_BACKLOG = int(os.environ.get("TQ_STORAGE_ZMQ_BACKLOG", 4096)) + +# Accept-queue sampling period in seconds; 0 disables the probe. Keep sub-second. +TQ_ACCEPT_PROBE_INTERVAL = float(os.environ.get("TQ_ACCEPT_PROBE_INTERVAL", 0)) + class StorageKeyNotFoundError(KeyError): """Raised when a requested global index is absent from a storage unit. @@ -168,6 +175,10 @@ class SimpleStorageUnit: zmq_server_info: ZMQ connection information for clients. """ + _requests_arrived = 0 + _arrivals_by_op: dict[str, int] = {} + _accept_probe = None + def __init__(self, storage_unit_size: int | None = None): """Initialize a SimpleStorageUnit with the specified size. @@ -180,6 +191,9 @@ def __init__(self, storage_unit_size: int | None = None): self.storage_data = StorageUnitData(self.storage_unit_size) + self._requests_arrived = 0 + self._arrivals_by_op = {} + # Internal communication address for proxy and workers self._inproc_addr = f"inproc://simple_storage_workers_{self.storage_unit_id}" @@ -206,6 +220,7 @@ def __init__(self, storage_unit_size: int | None = None): self.proxy_thread, self.zmq_context, self.put_get_socket, + self._accept_probe, ) def _init_zmq_socket(self) -> None: @@ -219,6 +234,7 @@ def _init_zmq_socket(self) -> None: # Frontend: ROUTER for receiving client requests self.put_get_socket = create_zmq_socket(self.zmq_context, zmq.ROUTER, self._node_ip) + self.put_get_socket.setsockopt(zmq.BACKLOG, TQ_STORAGE_ZMQ_BACKLOG) while True: try: @@ -229,6 +245,18 @@ def _init_zmq_socket(self) -> None: logger.warning(f"[{self.storage_unit_id}]: Try to bind ZMQ sockets failed, retrying...") continue + if TQ_ACCEPT_PROBE_INTERVAL > 0: + # Lazy: the probe shells out to ``ss`` on a timer, so keep it out of runs that + # have not enabled it. + from transfer_queue.utils.accept_probe import AcceptQueueProbe + + self._accept_probe = AcceptQueueProbe( + port=self._put_get_socket_port, + owner_id=str(self.storage_unit_id), + interval_s=TQ_ACCEPT_PROBE_INTERVAL, + ) + self._accept_probe.start() + # Backend: DEALER for worker communication (connected via zmq.proxy) self.worker_socket = create_zmq_socket(self.zmq_context, zmq.DEALER, self._node_ip) self.worker_socket.bind(self._inproc_addr) @@ -348,6 +376,9 @@ def _worker_routine(self) -> None: started = time.perf_counter() try: + self._requests_arrived += 1 + self._arrivals_by_op[operation.name] = self._arrivals_by_op.get(operation.name, 0) + 1 + logger.debug(f"[{self.storage_unit_id}]: worker received operation: {operation}") # Process request @@ -582,8 +613,23 @@ def _handle_get_metrics(self) -> ZMQMessage: "capacity": self.storage_unit_size, "active_keys": self.storage_data.active_key_count, "process_rss_bytes": process_rss, + # Counted on arrival; op_stats below only advances on completion. + "requests_arrived": self._requests_arrived, + "arrivals_by_op": dict(self._arrivals_by_op), } + if self._accept_probe is not None: + stats = self._accept_probe.stats + metrics["accept_queue"] = { + "backlog": stats.backlog, + "peak_recv_q": stats.peak_recv_q, + "peak_utilization": stats.peak_utilization, + "sk_drops_delta": stats.sk_drops_delta, + "listen_overflow_delta": stats.overflow_delta, + "non_overflow_drop_delta": stats.non_overflow_drop_delta, + "samples": stats.samples, + } + # Include per-operation stats if Prometheus metrics are enabled if self._metrics is not None: op_stats = {} @@ -741,6 +787,7 @@ def _shutdown_resources( proxy_thread: Thread | None, zmq_context: zmq.Context | None, put_get_socket: zmq.Socket | None, + accept_probe: "AcceptQueueProbe | None" = None, ) -> None: """Clean up resources on garbage collection.""" logger.info("Shutting down SimpleStorageUnit resources...") @@ -748,6 +795,10 @@ def _shutdown_resources( # Signal all threads to stop shutdown_event.set() + # Before the ZMQ teardown: the probe runs on its own timer and would outlive the unit. + if accept_probe is not None: + accept_probe.stop() + # Terminate put_get_socket if put_get_socket: put_get_socket.close(linger=0) diff --git a/transfer_queue/utils/accept_probe.py b/transfer_queue/utils/accept_probe.py new file mode 100644 index 00000000..916955c4 --- /dev/null +++ b/transfer_queue/utils/accept_probe.py @@ -0,0 +1,245 @@ +# Copyright 2025 Huawei Technologies Co., Ltd. All Rights Reserved. +# Copyright 2025 The TransferQueue Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Accept-queue sampling for the storage-unit ROUTER socket. + +A connection dropped while being established is invisible to both ends, so the only +way to see it is to read the kernel's own counters. Two caveats govern how callers +may read what this reports: ``ListenOverflows`` and ``ListenDrops`` are per network +namespace rather than per port, and the kernel charges a socket's ``sk_drops`` for +several establishment failures, of which a full queue is only one. See +``docs/metrics.md`` for the exported metrics. +""" + +from __future__ import annotations + +import re +import subprocess +import threading +import time +from dataclasses import dataclass + +from transfer_queue.utils.logging_utils import get_logger + +logger = get_logger(__name__) + +_RECVQ_RE = re.compile(r"^LISTEN\s+(\d+)\s+(\d+)\s+\S*:(\d+)\s", re.M) +_DROPS_RE = re.compile(r"\bd(\d+)\b") + + +@dataclass +class AcceptQueueSample: + """One reading of the accept queue for a single listening port.""" + + timestamp: float + recv_q: int + backlog: int + sk_drops: int + listen_overflows: int + listen_drops: int + + @property + def utilization(self) -> float: + """Queue depth as a fraction of the configured backlog.""" + return self.recv_q / self.backlog if self.backlog else 0.0 + + +@dataclass +class AcceptQueueStats: + """Peak and delta summary over a sampling window.""" + + port: int + samples: int = 0 + peak_recv_q: int = 0 + peak_utilization: float = 0.0 + backlog: int = 0 + first_sample: AcceptQueueSample | None = None + last_sample: AcceptQueueSample | None = None + + @property + def sk_drops_delta(self) -> int: + """Connections this socket dropped during the window, 0 until two samples exist.""" + if self.first_sample is None or self.last_sample is None: + return 0 + return self.last_sample.sk_drops - self.first_sample.sk_drops + + @property + def overflow_delta(self) -> int: + """Namespace-wide accept-queue overflows during the window, 0 until two samples exist.""" + if self.first_sample is None or self.last_sample is None: + return 0 + return self.last_sample.listen_overflows - self.first_sample.listen_overflows + + @property + def non_overflow_drop_delta(self) -> int: + """Establishment drops in this window that were not accept-queue overflows.""" + if self.first_sample is None or self.last_sample is None: + return 0 + drops = self.last_sample.listen_drops - self.first_sample.listen_drops + return drops - self.overflow_delta + + def describe(self) -> str: + """Return a one-line summary of the window, for the probe's shutdown log.""" + return ( + f"port={self.port} samples={self.samples} backlog={self.backlog} " + f"peak_recv_q={self.peak_recv_q} peak_util={self.peak_utilization:.1%} " + f"sk_drops_delta={self.sk_drops_delta} listen_overflow_delta={self.overflow_delta}" + ) + + +def _read_listen_socket(port: int) -> tuple[int, int, int] | None: + """Return (recv_q, backlog, sk_drops) for the listening socket on ``port``.""" + try: + out = subprocess.run(["ss", "-lntm"], capture_output=True, text=True, timeout=5, check=False).stdout + except (OSError, subprocess.SubprocessError) as exc: + logger.debug(f"accept-probe: ss failed: {exc}") + return None + + for match in _RECVQ_RE.finditer(out): + recv_q, backlog, found_port = (int(g) for g in match.groups()) + if found_port != port: + continue + # skmem lives on the continuation line right after the match. + tail = out[match.end() : match.end() + 400] + drops_match = _DROPS_RE.search(tail.split("\n")[1] if "\n" in tail else "") + return recv_q, backlog, int(drops_match.group(1)) if drops_match else 0 + return None + + +def _read_listen_overflows() -> tuple[int, int]: + """Return this namespace's (ListenOverflows, ListenDrops) from /proc/net/netstat.""" + try: + with open("/proc/net/netstat") as handle: + lines = handle.read().splitlines() + except OSError: + return 0, 0 + + for i, line in enumerate(lines): + if not line.startswith("TcpExt:") or "ListenOverflows" not in line: + continue + keys = line.split() + values = lines[i + 1].split() + try: + return ( + int(values[keys.index("ListenOverflows")]), + int(values[keys.index("ListenDrops")]), + ) + except (ValueError, IndexError): + return 0, 0 + return 0, 0 + + +def sample_accept_queue(port: int) -> AcceptQueueSample | None: + """Take one accept-queue reading for ``port``, or None if it cannot be read.""" + listen = _read_listen_socket(port) + if listen is None: + return None + recv_q, backlog, sk_drops = listen + overflows, drops = _read_listen_overflows() + return AcceptQueueSample( + timestamp=time.time(), + recv_q=recv_q, + backlog=backlog, + sk_drops=sk_drops, + listen_overflows=overflows, + listen_drops=drops, + ) + + +class AcceptQueueProbe: + """Sample one listening port's accept queue from a background thread. + + Args: + port: Listening port to watch. + owner_id: Identifier used in log lines (the storage unit id). + interval_s: Seconds between samples. Keep sub-second; the queue drains in + milliseconds, so a slower cadence misses the burst entirely. + warn_utilization: Warn once when depth first reaches this fraction of the backlog. + """ + + def __init__( + self, + port: int, + owner_id: str, + interval_s: float = 0.1, + warn_utilization: float = 0.5, + ) -> None: + self.port = port + self.owner_id = owner_id + self.interval_s = interval_s + self.warn_utilization = warn_utilization + self.stats = AcceptQueueStats(port=port) + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self._warned = False + + def start(self) -> None: + """Start the sampling thread; a second call is a no-op.""" + if self._thread is not None: + return + self._thread = threading.Thread(target=self._run, name=f"AcceptQueueProbe-{self.owner_id}", daemon=True) + self._thread.start() + logger.info(f"[{self.owner_id}]: accept-queue probe started on port {self.port} (interval={self.interval_s}s)") + + def stop(self) -> None: + """Stop the sampling thread and log the window summary.""" + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=5) + self._thread = None + logger.info(f"[{self.owner_id}]: accept-queue probe stopped. {self.stats.describe()}") + + def _run(self) -> None: + while not self._stop.is_set(): + sample = sample_accept_queue(self.port) + if sample is not None: + self._record(sample) + self._stop.wait(self.interval_s) + + def _record(self, sample: AcceptQueueSample) -> None: + stats = self.stats + stats.samples += 1 + stats.backlog = sample.backlog or stats.backlog + previous = stats.last_sample + if stats.first_sample is None: + stats.first_sample = sample + stats.last_sample = sample + + if sample.recv_q > stats.peak_recv_q: + stats.peak_recv_q = sample.recv_q + stats.peak_utilization = sample.utilization + + # Against the previous sample, not the first: sk_drops is cumulative, so measuring + # from probe start would re-report one old drop on every sample. + if previous is not None and sample.sk_drops > previous.sk_drops: + logger.error( + f"[{self.owner_id}]: listening socket on port {self.port} dropped an incoming " + f"connection. recv_q={sample.recv_q}/{sample.backlog} sk_drops={sample.sk_drops} " + f"(+{sample.sk_drops - previous.sk_drops} since the last sample, " + f"+{stats.sk_drops_delta} since probe start); netns since probe start: " + f"listen_overflows +{stats.overflow_delta}, other establishment drops " + f"+{stats.non_overflow_drop_delta}. The kernel charges sk_drops for several " + f"establishment failures, so a full queue is only one candidate: raising " + f"ZMQ_BACKLOG above {sample.backlog} helps only if recv_q above and the " + f"overflow delta point that way." + ) + + if not self._warned and sample.utilization >= self.warn_utilization: + self._warned = True + logger.warning( + f"[{self.owner_id}]: accept queue on port {self.port} reached " + f"{sample.recv_q}/{sample.backlog} ({sample.utilization:.0%}); " + f"connections are queueing faster than they are accepted." + )