feat(BA-5878): emit stats.rate window queries for io_read/io_write/net_rx/net_tx

Legacy live_stat consumers — most importantly the
legacy_compute_session.net_rx_bytes / net_tx_bytes / io_read_bytes /
io_write_bytes GraphQL resolvers — read the "stats.rate" key from the
per-metric live_stat dict. The new PromQL pipeline previously emitted
nothing under that label, leaving those four legacy fields uncovered:

- io_read / io_write: legacy agent's stats_filter={"rate"}, which the
  new pipeline did not produce.
- net_rx / net_tx: legacy agent's stats_filter is empty, so the agent
  publishes no stats.rate at all even though the legacy resolver
  expects it — making the resolver always return 0. The new pipeline
  now produces a value where legacy never did.

Two metric shapes flow through the new bucket:

- Gauge-shape (net_rx, net_tx): the metric's `current` value is
  already a per-second rate (set by agent's
  current_hook = lambda m: m.stats.rate), so PromQL only needs to
  sum across replicas and label_replace to "stats.rate".
- Counter-shape (io_read, io_write): the value is a cumulative byte
  counter, so PromQL applies rate(...[window]) before label_replace.

Live verified against Prometheus on a running kernel:

    {net_rx, stats.rate} = 27530
    {net_tx, stats.rate} = 30378
    {io_read, stats.rate} = 0
    {io_write, stats.rate} = 0

Re-introduces ValueType.RATE (and its to_legacy_live_stat_label /
from_live_stat_label round-trip) that was removed earlier in the
branch when no producer existed; the round-trip is now load-bearing
again.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: seedspirit

src/ai/backend/common/clients/prometheus/fixed_query_builder.py

@@ -12,6 +12,8 @@
     STATS_MAX_GAUGE_METRIC_PATTERNS,
     STATS_MAX_GAUGE_METRICS,
     STATS_MAX_OVER_RATE_METRICS,
+    STATS_RATE_COUNTER_METRICS,
+    STATS_RATE_GAUGE_METRICS,
     ContainerLiveStatQueries,
     ContainerMetricOptionalLabel,
     MetricType,
@@ -164,6 +166,32 @@ def get_container_live_stat_queries(
             rate=self._build_filtered_preset(kernel_ids, _RATE_LIVE_STAT_SPEC),
             max=self._build_window_stats_preset(kernel_ids, _MAX_STATS_BUCKET),
             avg=self._build_window_stats_preset(kernel_ids, _AVG_STATS_BUCKET),
+            rate_stats=self._build_rate_stats_preset(kernel_ids),
+        )
+
+    def _build_rate_stats_preset(
+        self,
+        kernel_ids: Sequence[KernelId],
+    ) -> MetricPreset:
+        kernel_id_regex = _regex_union([str(kid) for kid in kernel_ids])
+        group_by = ",".join(sorted(_LIVE_STAT_GROUP_BY))
+        stat_label = ValueType.RATE.to_legacy_live_stat_label()
+        parts: list[str] = []
+        if STATS_RATE_GAUGE_METRICS:
+            gauge_regex = _regex_union(sorted(STATS_RATE_GAUGE_METRICS))
+            selector = self._utilization_selector(kernel_id_regex, gauge_regex)
+            parts.append(self._labelled_sum(selector, group_by, stat_label))
+        if STATS_RATE_COUNTER_METRICS:
+            counter_regex = _regex_union(sorted(STATS_RATE_COUNTER_METRICS))
+            base = self._utilization_selector(kernel_id_regex, counter_regex)
+            selector = f"rate({base}[{self._timewindow}])"
+            parts.append(self._labelled_sum(selector, group_by, stat_label))
+        return MetricPreset(template=" or ".join(parts))
+
+    def _labelled_sum(self, selector: str, group_by: str, stat_label: str) -> str:
+        return (
+            f"label_replace(sum by ({group_by})({selector}),"
+            f'"value_type","{stat_label}","value_type",".*")'
         )
 
     def _build_window_stats_preset(

src/ai/backend/common/clients/prometheus/metric_types.py

@@ -61,16 +61,17 @@ class MetricType(StrEnum):
 
 @dataclass(frozen=True)
 class ContainerLiveStatQueries:
-    """Gauge / diff / rate / max / avg query preset bundle for container live stats."""
+    """Gauge / diff / rate / max / avg / rate_stats query preset bundle for container live stats."""
 
     gauge: MetricPreset
     diff: MetricPreset
     rate: MetricPreset
     max: MetricPreset
     avg: MetricPreset
+    rate_stats: MetricPreset
 
     def to_list(self) -> list[MetricPreset]:
-        return [self.gauge, self.diff, self.rate, self.max, self.avg]
+        return [self.gauge, self.diff, self.rate, self.max, self.avg, self.rate_stats]
 
 
 DIFF_METRICS: Final[frozenset[str]] = frozenset({"cpu_util"})
@@ -91,6 +92,16 @@ def to_list(self) -> list[MetricPreset]:
 STATS_MAX_OVER_RATE_METRICS: Final[frozenset[str]] = frozenset({"cpu_util"})
 STATS_AVG_OVER_RATE_METRICS: Final[frozenset[str]] = frozenset({"cpu_util"})
 
+# stats.rate emission targets the legacy stats.rate live_stat label.
+# Two metric shapes flow in:
+#   * "gauge" set: agent's current_hook already publishes per-second rate as
+#     the metric's `current` value, so we only need to sum across replicas
+#     and relabel to stats.rate (no PromQL rate() wrap).
+#   * "counter" set: the published series is a cumulative byte counter, so
+#     we apply rate(...[window]) to get bytes/sec before relabel.
+STATS_RATE_GAUGE_METRICS: Final[frozenset[str]] = frozenset({"net_rx", "net_tx"})
+STATS_RATE_COUNTER_METRICS: Final[frozenset[str]] = frozenset({"io_read", "io_write"})
+
 
 @dataclass
 class ContainerMetricResponseInfo:

src/ai/backend/common/clients/prometheus/types.py

@@ -10,6 +10,7 @@ class ValueType(StrEnum):
     PCT = "pct"
     MAX = "max"
     AVG = "avg"
+    RATE = "rate"
 
     @classmethod
     def from_live_stat_label(cls, value: str) -> "ValueType":
@@ -21,7 +22,7 @@ def from_live_stat_label(cls, value: str) -> "ValueType":
 
     def to_legacy_live_stat_label(self) -> str:
         match self:
-            case ValueType.MAX | ValueType.AVG:
+            case ValueType.MAX | ValueType.AVG | ValueType.RATE:
                 return f"stats.{self.value}"
             case _:
                 return self.value

tests/unit/manager/services/utilization_metric/test_container_metric.py

@@ -847,6 +847,29 @@ def test_stats_queries_render_legacy_labels_from_typed_value_types(self) -> None
             '"value_type","stats.avg","value_type",".*")'
         )
 
+    def test_rate_stats_query_renders_legacy_stats_rate_label(self) -> None:
+        kernel_id = KernelId(UUID("12345678-1234-5678-1234-567812345678"))
+        fixed_query_builder = FixedQueryBuilder("5m")
+
+        queries = fixed_query_builder.get_container_live_stat_queries([kernel_id])
+
+        assert queries.rate_stats.render() == (
+            "label_replace(sum by (container_metric_name,kernel_id,value_type)("
+            "backendai_container_utilization"
+            '{kernel_id=~"12345678-1234-5678-1234-567812345678",'
+            'container_metric_name=~"net_rx|net_tx",'
+            'value_type="current"}),'
+            '"value_type","stats.rate","value_type",".*")'
+            " or "
+            "label_replace(sum by (container_metric_name,kernel_id,value_type)(rate("
+            "backendai_container_utilization"
+            '{kernel_id=~"12345678-1234-5678-1234-567812345678",'
+            'container_metric_name=~"io_read|io_write",'
+            'value_type="current"}'
+            "[5m])),"
+            '"value_type","stats.rate","value_type",".*")'
+        )
+
 
 class TestKernelMetricValuesByKernel:
     def test_from_prometheus_response_maps_legacy_stat_label_to_value_type(self) -> None: