了解我们的Cloud功能在GCP中的状态

Question

我想知道云功能URL是否总是响应，重要的是要知道URL的状态并在GCP监控中显示。

是否有可能知道它是否活跃。如果它是可能的，谁可以帮助我与样本代码。

我正在尝试如下，

os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "sa.json"
cf_url = f'https://{region}-{self.cf_project_id}.cloudfunctions.net/{self.cf_name}'
var1=requests.get(cf_url)
print(var1.status_code)

我期待这个get调用应该给我状态码200，以知道CF URL是正常的。但是我得到的状态是403。

该服务帐户具有云功能查看器访问权限!

Answer 1

这是预期。您收到HTTP 403，因为您的请求没有被验证。

将GOOGLE_APPLICATION_CREDENTIALS变量设置为服务帐户将不会自动设置认证头。

除了你不需要的是Cloud Functions Invoker和Cloud Function Viewer扮演的角色。Cloud Function Viewer用于查看功能，不能触发功能

你可以试试这个答案:

from google.oauth2 import service_account
from google.auth.transport.requests import AuthorizedSession

url = 'https://test-123456.cloudfunctions.net/my-cloud-function'
creds = service_account.IDTokenCredentials.from_service_account_file(
'/path/to/service-account-credentials.json', target_audience=url)
authed_session = AuthorizedSession(creds)
# make authenticated request and print the response, status_code
resp = authed_session.get(url)
print(resp.status_code)
print(resp.text)

或者john Hanley在这里给出的代码:

import json
import base64
import requests
import google.auth.transport.requests
from google.oauth2.service_account import IDTokenCredentials
# The service account JSON key file to use to create the Identity Token
sa_filename = 'service-account.json'
# Endpoint to call
endpoint = 'https://us-east1-replace_with_project_id.cloudfunctions.net/main'
# The audience that this ID token is intended for (example Google Cloud Functions service URL)
aud = 'https://us-east1-replace_with_project_id.cloudfunctions.net/main'
def invoke_endpoint(url, id_token):
headers = {'Authorization': 'Bearer ' + id_token}
r = requests.get(url, headers=headers)
if r.status_code != 200:
print('Calling endpoint failed')
print('HTTP Status Code:', r.status_code)
print(r.content)
return None
return r.content.decode('utf-8')
if __name__ == '__main__':
credentials = IDTokenCredentials.from_service_account_file(
sa_filename,
target_audience=aud)
request = google.auth.transport.requests.Request()
credentials.refresh(request)
# This is debug code to show how to decode Identity Token
# print('Decoded Identity Token:')
# print_jwt(credentials.token.encode())
response = invoke_endpoint(endpoint, credentials.token)
if response is not None:
print(response)