嵌入式Linux设备Python上传Azure存储报403 TooManyDevicesError排查

在嵌入式Linux设备上运行Python 3.10.4应用，通过异步代码连接IoT Hub设备客户端，使用device_client_wrapper类配合信号量保证同一时间仅一个函数调用设备客户端方法。新增文件上传至Azure Blob Storage的函数后，初期运行正常，但多次执行后出现403 TooManyDevicesError错误，报错发生在device_client.get_storage_info_for_blob方法。已知单设备并发文件上传上限为10，每次仅上传7个文件且全部成功完成后再重试，信号量也限制了并发，怀疑存在资源未正确释放或未完成的情况。

报错栈信息 #

ERROR upload_file Exception: Unexpected failure
ERROR Traceback (most recent call last):
  File "/usr/lib/python3.10/site-packages/azure/iot/device/iothub/aio/async_clients.py", line 33, in handle_result
    return await callback.completion()
  File "/usr/lib/python3.10/site-packages/azure/iot/device/common/async_adapter.py", line 91, in completion
    return await self.future
azure.iot.device.exceptions.ServiceError: HTTP operation returned: 403 TooManyDevicesError(Error: Forbidden)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/root/cogo_ig60/cloud/device_client_wrapper.py", line 99, in upload_file
    blob_info = await self.device_client.get_storage_info_for_blob(blob_name)
  File "/usr/lib/python3.10/site-packages/azure/iot/device/iothub/aio/async_clients.py", line 571, in get_storage_info_for_blob
    storage_info = await handle_result(callback)
  File "/usr/lib/python3.10/site-packages/azure/iot/device/iothub/aio/async_clients.py", line 57, in handle_result
    raise exceptions.ClientError("Unexpected failure") from e
azure.iot.device.exceptions.ClientError: Unexpected failure

上传函数代码 #

async def upload_file(self, path_to_file: str, blob_name: str):
    """Upload a file to Azure Blob Storage."""
    try:
        async with self.semaphore:
            blob_info = await self.device_client.get_storage_info_for_blob(blob_name)
            sas_url = "https://{}/{}/{}{}".format(
                blob_info["hostName"],
                blob_info["containerName"], 
                blob_info["blobName"],
                blob_info["sasToken"]
            )
            logger.info(
                f"Uploading file {path_to_file} to blob storage as {blob_name}")
            with BlobClient.from_blob_url(sas_url) as blob_client:
                with open(path_to_file, "rb") as f:
                    result = blob_client.upload_blob(f, overwrite=True)
                    return (True, result)
    except FileNotFoundError as e:
        # catch file not found and add an HTTP status code to return in notification to IoT Hub
        logger.error(f'upload_file FileNotFoundError: {e}')
        logger.error(traceback.format_exc())
        e.status_code = 404
        return (False, e)
    except AzureError as e:
        # catch Azure errors that might result from the upload operation
        logger.error(f'upload_file AzureError: {e}')
        logger.error(traceback.format_exc())
        return (False, e)
    except Exception as e:
        # catch all other errors
        logger.error(f'upload_file Exception: {e}')
        logger.error(traceback.format_exc())
        return (False, e)

• 确认信号量作用范围
  检查self.semaphore初始化值是否设为7（或小于10），同时确保所有upload_file调用都复用同一个device_client_wrapper实例，避免多实例各自持信号量导致实际并发超限。

• 显式管理IoT Hub客户端连接
  上传完成后，检查客户端连接状态，异常时重置连接：

  # 上传完成后添加
  if not self.device_client.connected:
      await self.device_client.connect()
  

• 强化资源清理逻辑
  即使在with块中管理BlobClient，仍可显式调用blob_client.close()确保HTTP连接关闭；同时在异常分支中添加客户端断开重连逻辑，避免请求上下文泄漏：

  except Exception as e:
      logger.error(f'upload_file Exception: {e}')
      logger.error(traceback.format_exc())
      # 异常时尝试重置客户端连接
      try:
          if self.device_client.connected:
              await self.device_client.disconnect()
          await self.device_client.connect()
      except:
          pass
      return (False, e)
  

• 添加请求频率控制
  在每次上传完成后添加短暂异步延迟，避免短时间内请求过于密集触发服务端限制：

  # async with块内，上传完成后添加
  await asyncio.sleep(0.5)
  

• 升级Azure IoT SDK
  旧版本SDK可能存在异步资源泄漏bug，升级至最新稳定版：

  pip install --upgrade azure-iot-device
  

内容的提问来源于stack exchange，提问作者Locks