Соединение разорвано: IncompleteRead в Google Cloud Run

У меня есть служба запуска облака Google, которая запускается через pubsub, собирает данные из API Azure и помещает их в корзину gcp с помощью smart_open.

          with smart_open.open(destination, "wb") as fout:
            schema = pyarrow_adapters.json_schema_to_pyarrow_schema(self.desired_schema)
            data_frame.to_parquet(
                path=fout,
                engine = "pyarrow",
                compression = "gzip",
                schema=schema,
                use_compliant_nested_type=False,
            )

Редактировать: Это ошибка, которую я получаю в google-resumable-media-python

https://github.com/googleapis/google-resumable-media-python/blob/main/google/resumable_media/_upload.py#L50-L53

Это сохраняет сотни тысяч файлов за несколько часов, каждый день происходит несколько сбоев с этой ошибкой, которую я не могу понять, как исправить.

Получение метрик Azure для идентификатора ресурса: xxx Конфигурация: ххх Ошибка получения метрик: соединение разорвано: IncompleteRead(прочитано 15773 байт, ожидается еще 125469)', IncompleteRead(прочитано 15773 байт, ожидается еще 125469))

А также иногда очень похожая ошибка, подобная этой

Получение метрик Azure для идентификатора ресурса: xxx Конфигурация: ххх Ошибка получения метрик: поток байтов находится в неожиданном состоянии. Из локального потока было прочитано 8887 байт, а 0 байт уже были обновлены (они должны совпадать).

Метрики Cloudrun выглядят хорошо, и я понятия не имею, как их отлаживать дальше.

Шаги по повторению.

1. Настройте службу Cloudrun, которая запускает приложение Flask с одной конечной точкой, которая получает события от триггера eventarc.

   @app.route("/", methods=["POST"])
   def index():
     data = request.get_json()
     if not data:
       msg = "no Pub/Sub message received"
       print(f"error: {msg}")
       return f"Bad Request: {msg}", 400
   
     if not isinstance(data, dict) or "message" not in data:
       msg = "invalid Pub/Sub message formaty"
       print(f"error: {msg}")
       return f"Bad Request: {msg}", 400
   
     pubsub_message = data["message"]
   

2. Используйте Gunicorn, чтобы запустить это, Dockerfile

   ...
   CMD ./azu_metrics_extractor_pex.pex \
       azu_metrics_extractor.main:app \
       -b :${PORT} \
       -w 3 \
       --threads=8 \
       --timeout=0
   

3. Служба должна использовать информацию в сообщении pubsub для получения данных ресурсов из Azure, преобразования в паркет и записи их в корзину gcp с помощью smart_open.

         with smart_open.open(destination, "wb") as fout:
           schema = pyarrow_adapters.json_schema_to_pyarrow_schema(self.desired_schema)
           data_frame.to_parquet(
               path=fout,
               engine = "pyarrow",
               compression = "gzip",
               schema=schema,
               use_compliant_nested_type=False,
           )
   

4. Запустите тему pubsub с ~10 тыс. сообщений за час

5. Обратите внимание, что примерно 2 или 3 файла паркета, которые должны были быть сохранены в gcp, не удалось сохранить с указанными выше ошибками.

Потенциально связано с этой открытой проблемой для smart_open. https://github.com/piskvorky/smart_open/issues/784