Add detailed debug logs to topic instances

Author: vgvoleg

ydb/_topic_reader/topic_reader_asyncio.py

@@ -121,6 +121,7 @@ async def receive_batch(
 
         use asyncio.wait_for for wait with timeout.
         """
+        logger.debug("receive_batch max_messages=%s", max_messages)
         await self._reconnector.wait_message()
         return self._reconnector.receive_batch_nowait(
             max_messages=max_messages,
@@ -137,6 +138,7 @@ async def receive_batch_with_tx(
 
         use asyncio.wait_for for wait with timeout.
         """
+        logger.debug("receive_batch_with_tx tx=%s max_messages=%s", tx, max_messages)
         await self._reconnector.wait_message()
         return self._reconnector.receive_batch_with_tx_nowait(
             tx=tx,
@@ -149,6 +151,7 @@ async def receive_message(self) -> typing.Optional[datatypes.PublicMessage]:
 
         use asyncio.wait_for for wait with timeout.
         """
+        logger.debug("receive_message")
         await self._reconnector.wait_message()
         return self._reconnector.receive_message_nowait()
 
@@ -159,6 +162,7 @@ def commit(self, batch: typing.Union[datatypes.PublicMessage, datatypes.PublicBa
         For the method no way check the commit result
         (for example if lost connection - commits will not re-send and committed messages will receive again).
         """
+        logger.debug("commit message or batch")
         if self._settings.consumer is None:
             raise issues.Error("Commit operations are not supported for topic reader without consumer.")
 
@@ -177,6 +181,7 @@ async def commit_with_ack(self, batch: typing.Union[datatypes.PublicMessage, dat
         before receive commit ack. Message may be acked or not (if not - it will send in other read session,
         to this or other reader).
         """
+        logger.debug("commit_with_ack message or batch")
         if self._settings.consumer is None:
             raise issues.Error("Commit operations are not supported for topic reader without consumer.")
 
@@ -187,8 +192,10 @@ async def close(self, flush: bool = True):
         if self._closed:
             raise TopicReaderClosedError()
 
+        logger.debug("Close topic reader")
         self._closed = True
         await self._reconnector.close(flush)
+        logger.debug("Topic reader was closed")
 
     @property
     def read_session_id(self) -> Optional[str]:
@@ -219,6 +226,7 @@ def __init__(
         self._driver = driver
         self._loop = loop if loop is not None else asyncio.get_running_loop()
         self._background_tasks = set()
+        logger.debug("init reader reconnector id=%s", self._id)
 
         self._state_changed = asyncio.Event()
         self._stream_reader = None
@@ -231,13 +239,18 @@ async def _connection_loop(self):
         attempt = 0
         while True:
             try:
+                logger.debug("reader %s connect attempt %s", self._id, attempt)
                 self._stream_reader = await ReaderStream.create(self._id, self._driver, self._settings)
+                logger.debug(
+                    "reader %s connected stream %s", self._id, self._stream_reader._id
+                )
                 attempt = 0
                 self._state_changed.set()
                 await self._stream_reader.wait_error()
             except BaseException as err:
                 retry_info = check_retriable_error(err, self._settings._retry_settings(), attempt)
                 if not retry_info.is_retriable:
+                    logger.debug("reader %s stop connection loop due to %s", self._id, err)
                     self._set_first_error(err)
                     return
 
@@ -358,6 +371,7 @@ def commit(self, batch: datatypes.ICommittable) -> datatypes.PartitionSession.Co
         return self._stream_reader.commit(batch)
 
     async def close(self, flush: bool):
+        logger.debug("reader reconnector %s close", self._id)
         if self._stream_reader:
             await self._stream_reader.close(flush)
         for task in self._background_tasks:
@@ -447,6 +461,10 @@ def __init__(
 
         self._settings = settings
 
+        logger.debug(
+            "created ReaderStream id=%s reconnector=%s", self._id, self._reader_reconnector_id
+        )
+
     @staticmethod
     async def create(
         reader_reconnector_id: int,
@@ -464,6 +482,9 @@ async def create(
             get_token_function=creds.get_auth_token if creds else None,
         )
         await reader._start(stream, settings._init_message())
+        logger.debug(
+            "reader stream %s started session=%s", reader._id, reader._session_id
+        )
         return reader
 
     async def _start(self, stream: IGrpcWrapperAsyncIO, init_message: StreamReadMessage.InitRequest):
@@ -472,11 +493,17 @@ async def _start(self, stream: IGrpcWrapperAsyncIO, init_message: StreamReadMess
 
         self._started = True
         self._stream = stream
+        logger.debug(
+            "reader stream %s send init request", self._id
+        )
 
         stream.write(StreamReadMessage.FromClient(client_message=init_message))
         init_response = await stream.receive()  # type: StreamReadMessage.FromServer
         if isinstance(init_response.server_message, StreamReadMessage.InitResponse):
             self._session_id = init_response.server_message.session_id
+            logger.debug(
+                "reader stream %s initialized session=%s", self._id, self._session_id
+            )
         else:
             raise TopicReaderError("Unexpected message after InitRequest: %s", init_response)
 
@@ -615,6 +642,7 @@ async def _handle_background_errors(self):
 
     async def _read_messages_loop(self):
         try:
+            logger.debug("reader stream %s start read loop", self._id)
             self._stream.write(
                 StreamReadMessage.FromClient(
                     client_message=StreamReadMessage.ReadRequest(
@@ -628,6 +656,9 @@ async def _read_messages_loop(self):
                     _process_response(message.server_status)
 
                     if isinstance(message.server_message, StreamReadMessage.ReadResponse):
+                        logger.debug(
+                            "reader stream %s read %s bytes", self._id, message.server_message.bytes_size
+                        )
                         self._on_read_response(message.server_message)
 
                     elif isinstance(message.server_message, StreamReadMessage.CommitOffsetResponse):
@@ -637,18 +668,33 @@ async def _read_messages_loop(self):
                         message.server_message,
                         StreamReadMessage.StartPartitionSessionRequest,
                     ):
+                        logger.debug(
+                            "reader stream %s start partition %s",
+                            self._id,
+                            message.server_message.partition_session.partition_session_id,
+                        )
                         await self._on_start_partition_session(message.server_message)
 
                     elif isinstance(
                         message.server_message,
                         StreamReadMessage.StopPartitionSessionRequest,
                     ):
+                        logger.debug(
+                            "reader stream %s stop partition %s",
+                            self._id,
+                            message.server_message.partition_session_id,
+                        )
                         self._on_partition_session_stop(message.server_message)
 
                     elif isinstance(
                         message.server_message,
                         StreamReadMessage.EndPartitionSession,
                     ):
+                        logger.debug(
+                            "reader stream %s end partition %s",
+                            self._id,
+                            message.server_message.partition_session_id,
+                        )
                         self._on_end_partition_session(message.server_message)
 
                     elif isinstance(message.server_message, UpdateTokenResponse):
@@ -663,6 +709,7 @@ async def _read_messages_loop(self):
 
                 self._state_changed.set()
         except Exception as e:
+            logger.debug("reader stream %s error: %s", self._id, e)
             self._set_first_error(e)
             return
 
@@ -825,6 +872,9 @@ def _read_response_to_batches(self, message: StreamReadMessage.ReadResponse) ->
     async def _decode_batches_loop(self):
         while True:
             batch = await self._batches_to_decode.get()
+            logger.debug(
+                "reader stream %s decode batch %s messages", self._id, len(batch.messages)
+            )
             await self._decode_batch_inplace(batch)
             self._add_batch_to_queue(batch)
             self._state_changed.set()
@@ -833,9 +883,21 @@ def _add_batch_to_queue(self, batch: datatypes.PublicBatch):
         part_sess_id = batch._partition_session.id
         if part_sess_id in self._message_batches:
             self._message_batches[part_sess_id]._extend(batch)
+            logger.debug(
+                "reader stream %s extend batch partition=%s size=%s",
+                self._id,
+                part_sess_id,
+                len(batch.messages),
+            )
             return
 
         self._message_batches[part_sess_id] = batch
+        logger.debug(
+            "reader stream %s new batch partition=%s size=%s",
+            self._id,
+            part_sess_id,
+            len(batch.messages),
+        )
 
     async def _decode_batch_inplace(self, batch):
         if batch._codec == Codec.CODEC_RAW:
@@ -882,6 +944,7 @@ async def close(self, flush: bool):
             return
 
         self._closed = True
+        logger.debug("reader stream %s close", self._id)
 
         if flush:
             await self.flush()
@@ -899,3 +962,5 @@ async def close(self, flush: bool):
 
         if self._background_tasks:
             await asyncio.wait(self._background_tasks)
+
+        logger.debug("reader stream %s was closed", self._id)