生产环境下Google Cloud Vision API间歇性认证问题求助

我们部署在私有基础设施虚拟机上的FastAPI生产服务，调用Google Cloud Vision API进行文本检测时出现间歇性认证错误。服务通过服务账号JSON文件认证，使用Python的ImageAnnotatorClient实现功能，本地及开发服务器上运行正常，但生产环境中偶尔返回401认证错误，还偶发503服务不可用错误。

错误日志 #

401认证错误 #

2024-08-18 07:16:35,315 - uvicorn - ERROR - Response Message: 401 Request had invalid authentication credentials. Expected OAuth 2 access token, login cookie or other valid authentication credential. See https://developers.google.com/identity/sign-in/web/devconsole-project. [reason: "ACCESS_TOKEN_EXPIRED" domain: "googleapis.com" metadata {   key: "service"   value: "vision.googleapis.com" } metadata {   key: "method"   value: "google.cloud.vision.v1.ImageAnnotator.BatchAnnotateImages"

503服务不可用错误 #

gunicorn[3784527]: google.api_core.exceptions.ServiceUnavailable: 503 failed to connect to all addresses; last error: UNKNOWN: ipv4:142.250.181.74:443: Failed to connect to remote host: FD Shutdown

已排查动作 #

• 验证服务账号凭证及API配置正确，排除网络基础问题
• 尝试更换多个服务账号，问题仍存在
• 在生产服务器上直接运行Python客户端调用API时，每次均出现认证错误；但FastAPI服务内调用有时可正常工作

核心代码 #

def setup_client(self):
        main_key = Utils.main_key_compile(self.first_key_part_file,self.second_key_part_file,self.third_key_part_file)
        
        # Create a Fernet cipher using the main key
        cipher_suite = Fernet(main_key)

        with open(self.gvk_file_path, "rb") as file:
            encrypted_data = file.read()

        # Read the encrypted content from the JSON file you want to decrypt
        decrypted_data = cipher_suite.decrypt(encrypted_data)

        # client = vision.ImageAnnotatorClient()
        json_str = decrypted_data.decode("utf-8")  # Decode the bytes to a string
        api_key_dict = json.loads(json_str)  # Parse the JSON string into a Python dictionary
        my_credentials = service_account.Credentials.from_service_account_info(api_key_dict)
        self.client = vision.ImageAnnotatorClient(credentials=my_credentials)

    def extract_text_google(self, image):
        success, encoded_image = cv2.imencode('.png', image)
        content = encoded_image.tobytes()
        image = vision.Image(content=content)
        image_context = types.ImageContext(language_hints=["en"])
        max_retries = 2
        retry_delay = 2  # seconds
        for attempt in range(max_retries):
            try:
                response = self.client.text_detection(image=image,image_context=image_context)
                # check if response.text_annotations is empty   
                if len(response.text_annotations) == 0:
                    text_results = {}
                else:
                    text_results = response.text_annotations[0].description
                return text_results

            except google.api_core.exceptions.Unauthenticated as e:
                self.logger.error(f"Authentication error: {e}")
                # Optionally, reinitialize the client here if needed
                self.setup_client()

            except google.api_core.exceptions.GoogleAPIError as e:
                self.logger.error(f"API error: {e}")
                if attempt < max_retries:
                    time.sleep(retry_delay)
                else:
                    raise
            except Exception as e:
                self.logger.error(f"Unexpected error: {e}")
                if attempt < max_retries:
                    time.sleep(retry_delay)
                else:
                    raise

        return {}

求助内容 #

请求排查该间歇性认证问题的原因及解决方法。

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