Add topic reader received messages metric

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 3.32.1 ##
+* Add the `ydb.topic.reader.received.messages` counter and optional topic reader names for distinguishing reader metric series; bump the metrics build-info token to `ydb-sdk-metrics/0.2.0`
+
 ## 3.32.0 ##
 * Add `TableClient.read_rows` (sync and async) to read rows by primary key without a transaction
 * Add the `ydb.query.session.closed` counter for query session pool closures, labeled by pool name and a standardized closure reason; metrics-enabled clients now advertise `ydb-sdk-metrics/0.2.0` in `x-ydb-sdk-build-info`

docs/observability.rst

@@ -352,6 +352,9 @@ adapter maps them to instruments on the ``"ydb.sdk"`` meter):
    * - ``ydb.client.retry.attempts``
      - Histogram
      - Number of attempts performed for one logical retried operation.
+   * - ``ydb.topic.reader.received.messages``
+     - Counter (``{message}``)
+     - Messages accepted into the local SDK topic reader buffer.
 
 Attributes
 ~~~~~~~~~~
@@ -402,6 +405,15 @@ initial attach handshake does not count as closing an active session, and standa
 ``QuerySession`` instances do not publish pool metrics. A session publishes at most one
 closure event; the first terminal reason wins.
 
+``ydb.topic.reader.received.messages`` carries ``endpoint``, ``database``, ``topic``,
+``consumer``, and ``reader.name``. For reads without a consumer, ``consumer`` is an
+empty string. The user can pass ``reader_name`` to
+``TopicClient.reader``; otherwise the SDK generates a process-local ``reader-N`` value
+once for the logical reader and preserves it across reconnects. Use a rate function on
+this cumulative counter to diagnose incoming progress. A gap between received and an
+application-level delivered-message metric can indicate that the application is not
+consuming data or that decoding is failing.
+
 Writing a Custom Metrics Backend
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

docs/topic.rst

@@ -378,8 +378,14 @@ Reader Parameters
         consumer="my-consumer",
         buffer_size_bytes=50 * 1024 * 1024,  # client-side buffer (default: 50 MB)
         buffer_release_threshold=0.5,        # see below (default: 0.5)
+        reader_name="payments-worker",       # optional name used in reader metrics
     )
 
+``reader_name`` is an optional stable name for distinguishing topic readers in
+observability metrics. If it is omitted or empty, the SDK assigns a process-local name
+in the form ``reader-N``. Explicit names are not required to be unique: readers using
+the same name contribute to the same metric series when their other attributes match.
+
 ``buffer_size_bytes`` controls how many bytes the server is allowed to send before the client
 signals that it is ready for more. The server will not exceed this limit.
 

examples/opentelemetry/README.md

@@ -51,8 +51,11 @@ Grafana is provisioned with the **YDB Python SDK Metrics** dashboard. It uses
 Prometheus queries for SDK metrics such as `db_client_operation_duration`,
 `ydb_client_operation_failed`, `ydb_query_session_count`,
 `ydb_query_session_pending_requests`, `ydb_query_session_create_time`, and
-`ydb_client_retry_duration`. Use Grafana Explore for ad-hoc traces through Tempo
-and metrics through Prometheus.
+`ydb_client_retry_duration`. Topic readers also export the cumulative
+`ydb_topic_reader_received_messages_total` counter; use `rate(...)` and group by
+`topic`, `consumer`, or `reader_name` to inspect incoming progress (the OpenTelemetry
+attribute `reader.name` is normalized to `reader_name` by the Prometheus exporter). Use
+Grafana Explore for ad-hoc traces through Tempo and metrics through Prometheus.
 
 The SDK configures explicit OpenTelemetry histogram bucket boundaries for its
 own duration and retry-attempt metrics. Duration values are recorded in seconds,

tests/observability/test_metrics.py

@@ -103,10 +103,12 @@ def test_metrics_registry_records_all_instruments(metrics_setup, monkeypatch):
         QUERY_SESSION_TIMEOUTS,
         RETRY_ATTEMPTS,
         RETRY_DURATION,
+        TOPIC_READER_RECEIVED_MESSAGES,
         ATTEMPT_BUCKETS,
         DURATION_BUCKETS_SECONDS,
         RETRY_DURATION_BUCKETS_SECONDS,
         SessionMetrics,
+        TopicReaderMetrics,
         create_metrics_operation,
         record_query_session_count,
         record_query_session_create_time,
@@ -133,6 +135,11 @@ def test_metrics_registry_records_all_instruments(metrics_setup, monkeypatch):
     record_query_session_pending_requests(1, "main")
     record_query_session_timeout("main")
     record_retry_metrics(0.75, 3)
+    TopicReaderMetrics(
+        object(),
+        consumer_name=None,
+        reader_name="reader-test",
+    ).record_received_messages(1, "/Root/events")
 
     metrics = _metrics_by_name(metrics_setup)
 
@@ -148,6 +155,7 @@ def test_metrics_registry_records_all_instruments(metrics_setup, monkeypatch):
         QUERY_SESSION_TIMEOUTS,
         RETRY_ATTEMPTS,
         RETRY_DURATION,
+        TOPIC_READER_RECEIVED_MESSAGES,
     }
     assert metrics[CLIENT_OPERATION_DURATION].unit == "s"
     assert metrics[CLIENT_OPERATION_FAILED].unit == "{command}"
@@ -169,6 +177,7 @@ def test_metrics_registry_records_all_instruments(metrics_setup, monkeypatch):
         )
         assert _single_point_from_metrics(metrics, RETRY_DURATION).explicit_bounds == RETRY_DURATION_BUCKETS_SECONDS
         assert _single_point_from_metrics(metrics, RETRY_ATTEMPTS).explicit_bounds == ATTEMPT_BUCKETS
+    assert metrics[TOPIC_READER_RECEIVED_MESSAGES].unit == "{message}"
 
 
 def test_metrics_registry_supports_old_histogram_api():
@@ -294,6 +303,12 @@ def test_metrics_registry_is_noop_without_meter(monkeypatch):
     pool_metrics.close()
 
 
+def test_metrics_build_info_token_version(metrics_setup):
+    from ydb.observability import sdk_build_info_tokens
+
+    assert sdk_build_info_tokens() == ["ydb-sdk-metrics/0.2.0"]
+
+
 def test_metrics_operation_records_duration_once(metrics_setup, monkeypatch):
     from ydb.observability.metrics import CLIENT_OPERATION_DURATION, create_metrics_operation
 
@@ -1698,3 +1713,124 @@ def fake_span_ctx(**kwargs):
         assert qs._session_metrics._counted
         assert _single_point_for_pool(metrics_setup, QUERY_SESSION_COUNT, "async-session-pool").value == 1
         qs._close_session()
+
+
+def test_topic_reader_received_messages_attributes(metrics_setup):
+    from tests.observability.conftest import FakeDriverConfig
+    from ydb.observability.metrics import (
+        TOPIC_READER_RECEIVED_MESSAGES,
+        TopicReaderMetrics,
+    )
+
+    class FakeDriver:
+        _driver_config = FakeDriverConfig(
+            endpoint="grpc://localhost:2136",
+            database="/Root",
+        )
+
+    reader_metrics = TopicReaderMetrics(
+        FakeDriver(),
+        consumer_name="analytics",
+        reader_name="payments-worker",
+    )
+
+    reader_metrics.record_received_messages(
+        count=2,
+        topic="/Root/events",
+    )
+    reader_metrics.record_received_messages(
+        count=3,
+        topic="/Root/events",
+    )
+
+    point = _single_point(
+        metrics_setup,
+        TOPIC_READER_RECEIVED_MESSAGES,
+    )
+
+    assert point.value == 5
+    assert point.attributes == {
+        "endpoint": "localhost:2136",
+        "database": "/Root",
+        "topic": "/Root/events",
+        "consumer": "analytics",
+        "reader.name": "payments-worker",
+    }
+
+
+def test_topic_reader_received_messages_without_consumer(metrics_setup):
+    from ydb.observability.metrics import (
+        TOPIC_READER_RECEIVED_MESSAGES,
+        TopicReaderMetrics,
+    )
+
+    reader_metrics = TopicReaderMetrics(
+        object(),
+        consumer_name=None,
+        reader_name="reader-42",
+    )
+
+    reader_metrics.record_received_messages(
+        count=1,
+        topic="/Root/events",
+    )
+
+    point = _single_point(
+        metrics_setup,
+        TOPIC_READER_RECEIVED_MESSAGES,
+    )
+
+    assert point.attributes == {
+        "endpoint": "",
+        "database": "",
+        "topic": "/Root/events",
+        "consumer": "",
+        "reader.name": "reader-42",
+    }
+
+
+def test_topic_reader_received_messages_separates_topics(metrics_setup):
+    from ydb.observability.metrics import (
+        TOPIC_READER_RECEIVED_MESSAGES,
+        TopicReaderMetrics,
+    )
+
+    reader_metrics = TopicReaderMetrics(
+        object(),
+        consumer_name="analytics",
+        reader_name="worker",
+    )
+
+    reader_metrics.record_received_messages(2, "/Root/a")
+    reader_metrics.record_received_messages(3, "/Root/b")
+
+    values = {
+        point.attributes["topic"]: point.value
+        for point in _points(
+            metrics_setup,
+            TOPIC_READER_RECEIVED_MESSAGES,
+        )
+    }
+
+    assert values == {
+        "/Root/a": 2,
+        "/Root/b": 3,
+    }
+
+
+def test_topic_reader_received_messages_ignores_nonpositive_values(metrics_setup):
+    from ydb.observability.metrics import (
+        TOPIC_READER_RECEIVED_MESSAGES,
+        TopicReaderMetrics,
+    )
+
+    reader_metrics = TopicReaderMetrics(
+        object(),
+        consumer_name="analytics",
+        reader_name="worker",
+    )
+
+    reader_metrics.record_received_messages(0, "/Root/events")
+    reader_metrics.record_received_messages(-1, "/Root/events")
+
+    assert TOPIC_READER_RECEIVED_MESSAGES not in _metrics_by_name(metrics_setup)

ydb/_grpc/grpcwrapper/ydb_topic.py

@@ -486,11 +486,14 @@ class InitRequest(IToProto):
         topics_read_settings: List["StreamReadMessage.InitRequest.TopicReadSettings"]
         consumer: Optional[str]
         auto_partitioning_support: bool
+        reader_name: Optional[str] = None
 
         def to_proto(self) -> ydb_topic_pb2.StreamReadMessage.InitRequest:
             res = ydb_topic_pb2.StreamReadMessage.InitRequest()
             if self.consumer is not None:
                 res.consumer = self.consumer
+            if self.reader_name is not None:
+                res.reader_name = self.reader_name
             for settings in self.topics_read_settings:
                 res.topics_read_settings.append(settings.to_proto())
             res.auto_partitioning_support = self.auto_partitioning_support

ydb/_grpc/grpcwrapper/ydb_topic_test.py

@@ -2,7 +2,7 @@
 
 from google.protobuf.json_format import MessageToDict
 
-from ydb._grpc.grpcwrapper.ydb_topic import OffsetsRange
+from ydb._grpc.grpcwrapper.ydb_topic import OffsetsRange, StreamReadMessage
 from .ydb_topic import AlterTopicRequest
 from .ydb_topic_public_types import (
     AlterTopicRequestParams,
@@ -96,3 +96,27 @@ def test_alter_topic_request_from_public_to_proto():
     }
 
     assert msg_dict == expected_dict
+
+
+def test_stream_read_init_request_serializes_reader_name():
+    request = StreamReadMessage.InitRequest(
+        topics_read_settings=[],
+        consumer="analytics",
+        auto_partitioning_support=True,
+        reader_name="payments-worker",
+    )
+
+    proto = request.to_proto()
+
+    assert proto.reader_name == "payments-worker"
+
+
+def test_stream_read_init_request_omits_reader_name_by_default():
+    request = StreamReadMessage.InitRequest(
+        topics_read_settings=[],
+        consumer="analytics",
+        auto_partitioning_support=True,
+    )
+
+    assert request.reader_name is None
+    assert request.to_proto().reader_name == ""

ydb/_topic_reader/topic_reader.py

@@ -60,7 +60,15 @@ class PublicReaderSettings:
     buffer_release_threshold: float = 0.5
     """Min fraction of buffer_size_bytes to accumulate before sending a new ReadRequest (0.0 = immediately after every batch)."""
 
+    reader_name: Optional[str] = None
+    """Optional stable reader name used to distinguish reader metric series."""
+
     def __post_init__(self):
+        if self.reader_name is not None and not isinstance(
+            self.reader_name,
+            str,
+        ):
+            raise TypeError("Unsupported type for reader_name field: '%s'" % type(self.reader_name))
         if not (0.0 <= self.buffer_release_threshold <= 1.0):
             raise ValueError("buffer_release_threshold must be in [0.0, 1.0], got %s" % self.buffer_release_threshold)
         # check possible create init message
@@ -87,6 +95,7 @@ def _init_message(self) -> StreamReadMessage.InitRequest:
             topics_read_settings=list(map(PublicTopicSelector._to_topic_read_settings, selectors)),  # type: ignore
             consumer=self.consumer,
             auto_partitioning_support=self.auto_partitioning_support,
+            reader_name=self.reader_name,
         )
 
     def _retry_settings(self) -> RetrySettings: