René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Poor termination handling in  `bidi.BackgroundConsumer` resulting in error logs","articleBody":"## Overview\r\n\r\n* I have been trying to debug error logs that are popping up in our production infrastructure that I believe are \r\ncaused by poor termination handling logic in the `bidi.BackgroundConsumer` class.\r\n* The issues we are seeing occur when using the BigQuery Storage Write API, however I dont believe the underlying issue\r\nhas anything to do with the storage write client but instead is related to the `bidi.BackgroundConsumer` class.\r\n\r\n## Environment details\r\n\r\n  - OS type and version: Mac 14.2.1 (also occurs within google app engine flexible)\r\n  - Python version: 3.9.18\r\n  - pip version: 23.3.1\r\n  - `google-api-core` version: 2.17.1\r\n\r\n## Error 1: `bidi.BackgroundConsumer` error log on shutdown using very simple client server implementation\r\n\r\n* [simple_api_core_repro.py](https://github.com/perpetua1/tmp-grpc-client-and-bq-storage-write-errors-repro/blob/master/simple_api_core_repro.py): This script contains an extremely simple ping server and client \r\ncall using the `bidi.BidiRpc` and `bidi.BackgroundConsumer` classes which results in an error being logged.\r\n* The error is generated every single time the client is shutdown.\r\n\r\n\r\n#### Stack trace\r\n\r\n```\r\n14:38:29 - DEBUG - google.api_core.bidi - Thread-3 - Cleanly exiting request generator.\r\n14:38:29 - ERROR - google.api_core.bidi - Thread-ConsumeBidirectionalStream - Thread-ConsumeBidirectionalStream caught unexpected exception \u003c_MultiThreadedRendezvous of RPC that terminated with:\r\n\tstatus = StatusCode.CANCELLED\r\n\tdetails = \"Locally cancelled by application!\"\r\n\tdebug_error_string = \"None\"\r\n\u003e and will exit.\r\nTraceback (most recent call last):\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/google/api_core/bidi.py\", line 663, in _thread_main\r\n    response = self._bidi_rpc.recv()\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/google/api_core/bidi.py\", line 346, in recv\r\n    return next(self.call)\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/grpc/_channel.py\", line 540, in __next__\r\n    return self._next()\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/grpc/_channel.py\", line 966, in _next\r\n    raise self\r\ngrpc._channel._MultiThreadedRendezvous: \u003c_MultiThreadedRendezvous of RPC that terminated with:\r\n\tstatus = StatusCode.CANCELLED\r\n\tdetails = \"Locally cancelled by application!\"\r\n\tdebug_error_string = \"None\"\r\n\u003e\r\n14:38:29 - INFO - google.api_core.bidi - Thread-ConsumeBidirectionalStream - Thread-ConsumeBidirectionalStream exiting\r\n```\r\n\r\n## Error 2: Debug level exception in `bidi.BackgroundConsumer` on shutdown when using the BigQuery Storage Write API\r\n\r\n* See repro here: https://github.com/perpetua1/tmp-grpc-client-and-bq-storage-write-errors-repro/blob/master/bq_storage_write_repro.py\r\n* The same error from above is present in logs when using clients built on top of the BidiRpc/BackgroundConsumer like the\r\n  BigQuery Storage Write API but it shows up differently.\r\n* Specifically, something related to how gRPC channels are created in official google clients means that the channel\r\n  gets wrapped in a `_StreamingResponseIterator` which converts the `_MultiThreadedRendezvous` into a `GoogleAPICallError`\r\n* This means the level is reduced to DEBUG but a stack trace is still present in the logs.\r\n\r\n#### Stack trace\r\n\r\n```\r\n...\r\n14:41:08 - DEBUG - google.api_core.bidi - MainThread - Started helper thread Thread-ConsumeBidirectionalStream\r\n14:41:13 - DEBUG - google.auth.transport.requests - Thread-3 - Making request: POST https://oauth2.googleapis.com/token\r\n14:41:13 - DEBUG - urllib3.connectionpool - Thread-3 - Starting new HTTPS connection (1): oauth2.googleapis.com:443\r\n14:41:13 - DEBUG - urllib3.connectionpool - Thread-3 - https://oauth2.googleapis.com:443 \"POST /token HTTP/1.1\" 200 None\r\n14:41:14 - DEBUG - google.api_core.bidi - Thread-ConsumeBidirectionalStream - waiting for recv.\r\n14:41:14 - DEBUG - google.api_core.bidi - Thread-ConsumeBidirectionalStream - recved response.\r\n14:41:14 - DEBUG - google.api_core.bidi - Thread-ConsumeBidirectionalStream - waiting for recv.\r\n14:41:14 - DEBUG - google.cloud.bigquery_storage_v1.writer - MainThread - Stopping consumer.\r\n14:41:14 - DEBUG - google.api_core.bidi - Thread-2 - Cleanly exiting request generator.\r\n14:41:14 - DEBUG - google.api_core.bidi - Thread-ConsumeBidirectionalStream - Thread-ConsumeBidirectionalStream caught error 499 Locally cancelled by application! and will exit. Generally this is due to the RPC itself being cancelled and the error will be surfaced to the calling code.\r\nTraceback (most recent call last):\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/google/api_core/grpc_helpers.py\", line 116, in __next__\r\n    return next(self._wrapped)\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/grpc/_channel.py\", line 540, in __next__\r\n    return self._next()\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/grpc/_channel.py\", line 966, in _next\r\n    raise self\r\ngrpc._channel._MultiThreadedRendezvous: \u003c_MultiThreadedRendezvous of RPC that terminated with:\r\n\tstatus = StatusCode.CANCELLED\r\n\tdetails = \"Locally cancelled by application!\"\r\n\tdebug_error_string = \"None\"\r\n\u003e\r\n\r\nThe above exception was the direct cause of the following exception:\r\n\r\nTraceback (most recent call last):\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/google/api_core/bidi.py\", line 663, in _thread_main\r\n    response = self._bidi_rpc.recv()\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/google/api_core/bidi.py\", line 346, in recv\r\n    return next(self.call)\r\n  File \"/Users/dhendry/code/tmp-grpc-client-and-bq-storage-write-errors-repro/.venv/lib/python3.9/site-packages/google/api_core/grpc_helpers.py\", line 119, in __next__\r\n    raise exceptions.from_grpc_error(exc) from exc\r\ngoogle.api_core.exceptions.Cancelled: 499 Locally cancelled by application!\r\n14:41:14 - INFO - google.cloud.bigquery_storage_v1.writer - Thread-1 - RPC termination has signaled streaming pull manager shutdown.\r\n14:41:14 - INFO - google.api_core.bidi - Thread-ConsumeBidirectionalStream - Thread-ConsumeBidirectionalStream exiting\r\n14:41:14 - DEBUG - google.cloud.bigquery_storage_v1.writer - MainThread - Finished stopping manager.\r\n14:41:14 - DEBUG - root - MainThread - Stream closed, insertion success\r\n```\r\n\r\n## Error 3: Uncaught exception in `bidi.BackgroundConsumer` log when using the BigQuery Storage Write API\r\n\r\n* This is the primary reason I am opening this bug\r\n* I have not been able to craft a reproducible example for this error, however the relevant stack trace is below and this\r\n occurs in our production environment (Google AppEngine flexible) 10-100 times per day.\r\n* It seems to be occurring along a very similar code path to the above example however the internal state of the grpc\r\n  channel seems to be a bit different and the result is an uncaught `StopIteration` exception\r\n* It gets logged as an error.\r\n\r\n#### Stack trace\r\n\r\n```\r\nThread-ConsumeBidirectionalStream caught unexpected exception  and will exit.\r\nTraceback (most recent call last):\r\n  File \"/.venv/lib/python3.9/site-packages/google/api_core/bidi.py\", line 663, in _thread_main\r\n    response = self._bidi_rpc.recv()\r\n  File \"/.venv/lib/python3.9/site-packages/google/api_core/bidi.py\", line 346, in recv\r\n    return next(self.call)\r\n  File \"/.venv/lib/python3.9/site-packages/google/api_core/grpc_helpers.py\", line 119, in __next__\r\n    return next(self._wrapped)\r\n  File \"/.venv/lib/python3.9/site-packages/ddtrace/contrib/grpc/client_interceptor.py\", line 162, in __next__\r\n    return self._next()\r\n  File \"/.venv/lib/python3.9/site-packages/ddtrace/contrib/grpc/client_interceptor.py\", line 144, in _next\r\n    return next(self.__wrapped__)\r\n  File \"/.venv/lib/python3.9/site-packages/grpc/_channel.py\", line 540, in __next__\r\n    return self._next()\r\n  File \"/.venv/lib/python3.9/site-packages/grpc/_channel.py\", line 964, in _next\r\n    raise StopIteration()\r\nStopIteration\r\n```\r\n\r\n## Proposed fix\r\n\r\nI think just a bit of extra logic in the `bidi.BackgroundConsumer` class to catch and handle the `StopIteration` and \r\ncanceled exceptions would be sufficient to fix this issue in the second and third examples above. Specifically updating\r\nthe `google.api_core.bidi.BackgroundConsumer._thread_main()` implementation with the following block:\r\n\r\n```python\r\n        except (exceptions.Cancelled, StopIteration):\r\n            pass\r\n```\r\n\r\nMoreover catching the _MultiThreadedRendezvous could be done as well although this is admittedly very strange.\r\n\r\n```python\r\n        except _MultiThreadedRendezvous as exc:\r\n            if exc.code() != grpc.StatusCode.CANCELLED:\r\n                _LOGGER.exception(\r\n                    \"%s caught unexpected exception %s and will exit.\",\r\n                    _BIDIRECTIONAL_CONSUMER_NAME,\r\n                    exc,\r\n                )\r\n```\r\n\r\nA better approach could be to update the `grpc._channel._MultiThreadedRendezvous._next()` implemenation to raise\r\n`StopIteration` on `StatusCode.CANCELLED` in addition to `OK`.\r\n\r\nSo the full, the contents of `bidi.BackgroundConsumer`\r\n\r\n```python\r\n    def _thread_main(self, ready):\r\n        try:\r\n            ready.set()\r\n            self._bidi_rpc.add_done_callback(self._on_call_done)\r\n            self._bidi_rpc.open()\r\n\r\n            while self._bidi_rpc.is_active:\r\n                # Do not allow the paused status to change at all during this\r\n                # section. There is a condition where we could be resumed\r\n                # between checking if we are paused and calling wake.wait(),\r\n                # which means that we will miss the notification to wake up\r\n                # (oops!) and wait for a notification that will never come.\r\n                # Keeping the lock throughout avoids that.\r\n                # In the future, we could use `Condition.wait_for` if we drop\r\n                # Python 2.7.\r\n                # See: https://github.com/googleapis/python-api-core/issues/211\r\n                with self._wake:\r\n                    while self._paused:\r\n                        _LOGGER.debug(\"paused, waiting for waking.\")\r\n                        self._wake.wait()\r\n                        _LOGGER.debug(\"woken.\")\r\n\r\n                _LOGGER.debug(\"waiting for recv.\")\r\n                response = self._bidi_rpc.recv()\r\n                _LOGGER.debug(\"recved response.\")\r\n                self._on_response(response)\r\n        ######################################## \r\n        # ADD THIS:\r\n        except (exceptions.Cancelled, StopIteration):\r\n            pass\r\n        # AND MAYBE THIS, although its weird, perhaps there is a better way:\r\n        except _MultiThreadedRendezvous as exc:\r\n            if exc.code() != grpc.StatusCode.CANCELLED:\r\n                _LOGGER.exception(\r\n                    \"%s caught unexpected exception %s and will exit.\",\r\n                    _BIDIRECTIONAL_CONSUMER_NAME,\r\n                    exc,\r\n                )\r\n        ######################################## \r\n        except exceptions.GoogleAPICallError as exc:\r\n            _LOGGER.debug(\r\n                \"%s caught error %s and will exit. Generally this is due to \"\r\n                \"the RPC itself being cancelled and the error will be \"\r\n                \"surfaced to the calling code.\",\r\n                _BIDIRECTIONAL_CONSUMER_NAME,\r\n                exc,\r\n                exc_info=True,\r\n            )\r\n\r\n        except Exception as exc:\r\n            _LOGGER.exception(\r\n                \"%s caught unexpected exception %s and will exit.\",\r\n                _BIDIRECTIONAL_CONSUMER_NAME,\r\n                exc,\r\n            )\r\n\r\n        _LOGGER.info(\"%s exiting\", _BIDIRECTIONAL_CONSUMER_NAME)\r\n```\r\n","author":{"url":"https://github.com/dhendry","@type":"Person","name":"dhendry"},"datePublished":"2024-02-21T20:43:13.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":5},"url":"https://github.com/15057/google-cloud-python/issues/15057"}