3

我有一种情况，我正在尝试创建两个云功能，即 CF1 和 CF2，并且我有一个云调度程序。两个云功能都启用了经过身份验证的调用。我的流程是 Cloud Scheduler 将触发 CF1。CF1 完成后，CF1 将触发 CF2 作为 http 调用。我已经提到无法从 GCP 调度程序调用 Google Cloud Function以从 Cloud Scheduler 访问经过身份验证的 CF1 并能够访问 CF1。但是从 CF1 访问 CF2 时出现问题。CF1 不会触发 CF2，也不会给出任何错误消息。从另一个经过身份验证的 Cloud Function 访问经过身份验证的 Cloud Function 时，我们是否需要遵循任何其他技术。

CF1 代码：

import json
import logging
from requests_futures.sessions import FuturesSession


def main(request):
    # To read parameter values from request (url arguments or Json body).
    raw_request_data = request.data
    string_request_data = raw_request_data.decode("utf-8")
    request_json: dict = json.loads(string_request_data)

    request_args = request.args

    if request_json and 'cf2_endpoint' in request_json:
        cf2_endpoint = request_json['cf2_endpoint']
    elif request_args and 'cf2_endpoint' in request_args:
        cf2_endpoint = request_args['cf2_endpoint']
    else:
        cf2_endpoint = 'Invalid endpoint for CF2'

    logger = logging.getLogger('test')
    try:
        session = FuturesSession()
        session.get("{}".format(cf2_endpoint))
        logger.info("First cloud function executed successfully.")

    except RuntimeError:
        logger.error("Exception occurred {}".format(RuntimeError))

CF2 代码：

import logging

def main(request):
    logger = logging.getLogger('test')
    logger.info("second cloud function executed successfully.")

当前输出日志：

First cloud function executed successfully.

预期的输出日志：

First cloud function executed successfully.
second cloud function executed successfully.

注意：如果我对这两个云功能使用未经身份验证的访问，则相同的流程正在工作。

4

2 回答 2

0

Google Cloud Function 提供REST API 接口，其中包括可在另一个 Cloud Function HTTP 调用中使用的调用方法。尽管文档中提到了使用 Google 提供的客户端库，但 Python 上的 Cloud Function 仍然没有。

相反，您需要使用通用的 Google API 客户端库。[这是蟒蛇之一]。3

使用这种方法的主要困难可能是对身份验证过程的理解。通常，您需要提供两件事来构建客户端服务： 凭据和范围。

获取凭据的最简单方法是在应用程序默认凭据 (ADC) 库上进行中继。关于此的正确文档是：

  1. https://cloud.google.com/docs/authentication/production
  2. https://github.com/googleapis/google-api-python-client/blob/master/docs/auth.md

获取范围的地方是每个 REST API 函数文档页面。比如，OAuth 范围：https://www.googleapis.com/auth/cloud-platform

调用“hello-world”云函数的完整代码示例如下。运行前：

  1. 在您的项目中在 GCP 上创建默认 Cloud Function。
  • 保留并注意要使用的默认服务帐户
  • 保留默认正文。
  1. 注意project_id、函数名称、部署函数的位置。
  2. 如果您将在 Cloud Function 环境（例如本地）之外调用函数，请根据上述文档设置环境变量 GOOGLE_APPLICATION_CREDENTIALS
  3. 如果您实际上是从另一个 Cloud Function 调用，则根本不需要配置凭据。
from googleapiclient.discovery import build
from googleapiclient.discovery_cache.base import Cache
import google.auth

import pprint as pp

def get_cloud_function_api_service():
    class MemoryCache(Cache):
        _CACHE = {}

        def get(self, url):
            return MemoryCache._CACHE.get(url)

        def set(self, url, content):
            MemoryCache._CACHE[url] = content

    scopes = ['https://www.googleapis.com/auth/cloud-platform']

    # If the environment variable GOOGLE_APPLICATION_CREDENTIALS is set,
    # ADC uses the service account file that the variable points to.
    #
    # If the environment variable GOOGLE_APPLICATION_CREDENTIALS isn't set,
    # ADC uses the default service account that Compute Engine, Google Kubernetes Engine, App Engine, Cloud Run,
    # and Cloud Functions provide
    #
    # see more on https://cloud.google.com/docs/authentication/production
    credentials, project_id = google.auth.default(scopes)

    service = build('cloudfunctions', 'v1', credentials=credentials, cache=MemoryCache())
    return service


google_api_service = get_cloud_function_api_service()
name = 'projects/{project_id}/locations/us-central1/functions/function-1'
body = {
    'data': '{ "message": "It is awesome, you are develop on Stack Overflow language!"}' # json passed as a string
}
result_call = google_api_service.projects().locations().functions().call(name=name, body=body).execute()
pp.pprint(result_call)
# expected out out is:
# {'executionId': '3h4c8cb1kwe2', 'result': 'It is awesome, you are develop on Stack Overflow language!'}
于 2020-09-18T16:16:37.400 回答
0

这里发生了两件事：

  1. 您没有request-futures完全正确使用。由于请求是异步发出的，因此您需要在函数隐式返回之前阻止结果，否则它可能会在您的 HTTP 请求完成之前返回（尽管它可能在此示例中）：
session = FuturesSession()
future = session.get("{}".format(cf2_endpoint))
resp = future.result()  # Block on the request completing
  1. 您对第二个函数发出的请求实际上并不是经过身份验证的请求。默认情况下，来自云函数的出站请求未经过身份验证。如果您查看上面的实际响应，您会看到：
>>> resp.status_code
403
>>> resp.content
b'\n<html><head>\n<meta http-equiv="content-type" content="text/html;charset=utf-8">\n<title>403 Forbidden</title>\n</head>\n<body text=#000000 bgcolor=#ffffff>\n<h1>Error: Forbidden</h1>\n<h2>Your client does not have permission to get URL <code>/function_two</code> from this server.</h2>\n<h2></h2>\n</body></html>\n'

您可以跳过很多麻烦来正确验证此请求，如文档中所述：https ://cloud.google.com/functions/docs/securing/authenticating#function-to-function

但是，更好的选择是将您的第二个函数设为“背景”函数，并通过从第一个函数发布的 PubSub 消息调用它：

from google.cloud import pubsub

publisher = pubsub.PublisherClient()
topic_name = 'projects/{project_id}/topics/{topic}'.format(
    project_id=<your project id>,
    topic='MY_TOPIC_NAME',  # Set this to something appropriate.
)

def function_one(request):
    message = b'My first message!'
    publisher.publish(topic_name, message)

def function_two(event, context):
    message = event['data'].decode('utf-8')
    print(message)

只要您的函数具有发布 PubSub 消息的权限，这就避免了对 HTTP 请求添加授权的需要，并且还确保了至少一次交付。

于 2020-04-07T16:08:58.450 回答