adjust message logic

Author: abbrowne126

google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py

@@ -60,6 +60,12 @@
 
 
 _LOGGER = logging.getLogger(__name__)
+_SLOW_ACK_LOGGER = logging.getLogger("slow-ack")
+_STREAMS_LOGGER = logging.getLogger("subscriber-streams")
+_FLOW_CONTROL_LOGGER = logging.getLogger("subscriber-flow-control")
+_CALLBACK_DELIVERY_LOGGER = logging.getLogger("callback-delivery")
+_CALLBACK_EXCEPTION_LOGGER = logging.getLogger("callback-exceptions")
+_EXPIRY_LOGGER = logging.getLogger("expiry")
 _REGULAR_SHUTDOWN_THREAD_NAME = "Thread-RegularStreamShutdown"
 _RPC_ERROR_THREAD_NAME = "Thread-OnRpcTerminated"
 _RETRYABLE_STREAM_ERRORS = (
@@ -145,6 +151,14 @@ def _wrap_callback_errors(
         callback: The user callback.
         message: The Pub/Sub message.
     """
+    _CALLBACK_DELIVERY_LOGGER.debug(
+        "Message (id=%s, ack_id=%s, ordering_key=%s, exactly_once=%s) received by subscriber callback",
+        message.message_id,
+        message.ack_id,
+        message.ordering_key,
+        message.exactly_once_enabled,
+    )
+
     try:
         if message.opentelemetry_data:
             message.opentelemetry_data.end_subscribe_concurrency_control_span()
@@ -156,9 +170,15 @@ def _wrap_callback_errors(
         # Note: the likelihood of this failing is extremely low. This just adds
         # a message to a queue, so if this doesn't work the world is in an
         # unrecoverable state and this thread should just bail.
-        _LOGGER.exception(
-            "Top-level exception occurred in callback while processing a message"
+
+        _CALLBACK_EXCEPTION_LOGGER.exception(
+            "Message (id=%s, ack_id=%s, ordering_key=%s, exactly_once=%s)'s callback threw exception, nacking message.",
+            message.message_id,
+            message.ack_id,
+            message.ordering_key,
+            message.exactly_once_enabled,
         )
+
         message.nack()
         on_callback_error(exc)
 
@@ -199,6 +219,7 @@ def _process_requests(
     error_status: Optional["status_pb2.Status"],
     ack_reqs_dict: Dict[str, requests.AckRequest],
     errors_dict: Optional[Dict[str, str]],
+    ack_histogram: Optional[histogram.Histogram] = None,
 ):
     """Process requests when exactly-once delivery is enabled by referring to
     error_status and errors_dict.
@@ -210,6 +231,16 @@ def _process_requests(
     requests_completed = []
     requests_to_retry = []
     for ack_id in ack_reqs_dict:
+        # Debug logging: slow acks
+        if ack_histogram and ack_reqs_dict[
+            ack_id
+        ].time_to_ack > ack_histogram.percentile(percent=99):
+            _SLOW_ACK_LOGGER.debug(
+                "Message (id=%s, ack_id=%s) ack duration of %s s is higher than the p99 ack duration",
+                ack_reqs_dict[ack_id].message_id,
+                ack_reqs_dict[ack_id].ack_id,
+            )
+
         # Handle special errors returned for ack/modack RPCs via the ErrorInfo
         # sidecar metadata when exactly-once delivery is enabled.
         if errors_dict and ack_id in errors_dict:
@@ -560,8 +591,10 @@ def maybe_pause_consumer(self) -> None:
         with self._pause_resume_lock:
             if self.load >= _MAX_LOAD:
                 if self._consumer is not None and not self._consumer.is_paused:
-                    _LOGGER.debug(
-                        "Message backlog over load at %.2f, pausing.", self.load
+                    _FLOW_CONTROL_LOGGER.debug(
+                        "Message backlog over load at %.2f (threshold %.2f), initiating client-side flow control",
+                        self.load,
+                        _RESUME_THRESHOLD,
                     )
                     self._consumer.pause()
 
@@ -588,10 +621,18 @@ def maybe_resume_consumer(self) -> None:
             self._maybe_release_messages()
 
             if self.load < _RESUME_THRESHOLD:
-                _LOGGER.debug("Current load is %.2f, resuming consumer.", self.load)
+                _FLOW_CONTROL_LOGGER.debug(
+                    "Current load is %.2f (threshold %.2f), suspending client-side flow control.",
+                    self.load,
+                    _RESUME_THRESHOLD,
+                )
                 self._consumer.resume()
             else:
-                _LOGGER.debug("Did not resume, current load is %.2f.", self.load)
+                _FLOW_CONTROL_LOGGER.debug(
+                    "Current load is %.2f (threshold %.2f), retaining client-side flow control.",
+                    self.load,
+                    _RESUME_THRESHOLD,
+                )
 
     def _maybe_release_messages(self) -> None:
         """Release (some of) the held messages if the current load allows for it.
@@ -702,7 +743,7 @@ def send_unary_ack(
 
         if self._exactly_once_delivery_enabled():
             requests_completed, requests_to_retry = _process_requests(
-                error_status, ack_reqs_dict, ack_errors_dict
+                error_status, ack_reqs_dict, ack_errors_dict, self.ack_histogram
             )
         else:
             requests_completed = []
@@ -796,7 +837,7 @@ def send_unary_modack(
 
         if self._exactly_once_delivery_enabled():
             requests_completed, requests_to_retry = _process_requests(
-                error_status, ack_reqs_dict, modack_errors_dict
+                error_status, ack_reqs_dict, modack_errors_dict, self.ack_histogram
             )
         else:
             requests_completed = []
@@ -1239,6 +1280,11 @@ def _on_response(self, response: gapic_types.StreamingPullResponse) -> None:
             receipt_modack=True,
         )
 
+        if len(expired_ack_ids):
+            _EXPIRY_LOGGER.debug(
+                "ack ids %s were dropped as they have already expired."
+            )
+
         with self._pause_resume_lock:
             if self._scheduler is None or self._leaser is None:
                 _LOGGER.debug(
@@ -1304,9 +1350,13 @@ def _should_recover(self, exception: BaseException) -> bool:
         # If this is in the list of idempotent exceptions, then we want to
         # recover.
         if isinstance(exception, _RETRYABLE_STREAM_ERRORS):
-            _LOGGER.debug("Observed recoverable stream error %s", exception)
+            _STREAMS_LOGGER.debug(
+                "Observed recoverable stream error %s, reopening stream", exception
+            )
             return True
-        _LOGGER.debug("Observed non-recoverable stream error %s", exception)
+        _STREAMS_LOGGER.debug(
+            "Observed non-recoverable stream error %s, shutting down stream", exception
+        )
         return False
 
     def _should_terminate(self, exception: BaseException) -> bool:
@@ -1326,9 +1376,13 @@ def _should_terminate(self, exception: BaseException) -> bool:
         is_api_error = isinstance(exception, exceptions.GoogleAPICallError)
         # Terminate any non-API errors, or non-retryable errors (permission denied, unauthorized, etc.)
         if not is_api_error or isinstance(exception, _TERMINATING_STREAM_ERRORS):
-            _LOGGER.debug("Observed terminating stream error %s", exception)
+            _STREAMS_LOGGER.debug(
+                "Observed terminating stream error %s, shutting down stream", exception
+            )
             return True
-        _LOGGER.debug("Observed non-terminating stream error %s", exception)
+        _STREAMS_LOGGER.debug(
+            "Observed non-terminating stream error %s, attempting to reopen", exception
+        )
         return False
 
     def _on_rpc_done(self, future: Any) -> None:

google/cloud/pubsub_v1/subscriber/message.py

@@ -133,7 +133,7 @@ def __init__(
         self._delivery_attempt = delivery_attempt if delivery_attempt > 0 else None
         self._request_queue = request_queue
         self._exactly_once_delivery_enabled_func = exactly_once_delivery_enabled_func
-        self._message_id = message.message_id
+        self.message_id = message.message_id
 
         # The instantiation time is the time that this message
         # was received. Tracking this provides us a way to be smart about
@@ -231,11 +231,6 @@ def ack_id(self) -> str:
         """the ID used to ack the message."""
         return self._ack_id
 
-    @property
-    def message_id(self) -> str:
-        """The message id of the message"""
-        return self._message_id
-
     @property
     def delivery_attempt(self) -> Optional[int]:
         """The delivery attempt counter is 1 + (the sum of number of NACKs