借助 Cloud Functions 函数任务，您可以从集成中配置和运行 Google Cloud Functions。Google Cloud Functions 是一个轻量级计算解决方案，可供开发者创建单一用途的独立函数，无需管理服务器或运行时环境即可对 Cloud 事件作出响应。
如需了解详情，请参阅 Google Cloud Functions 文档。
须知事项
在配置 Cloud Functions 函数任务之前，请确保在您的 Google Cloud 项目中执行以下任务。
- 启用 Cloud Functions API (
cloudfunctions.googleapis.com)。 -
将以下 IAM 角色分配给您的主账号：
- Cloud Functions Admin (
roles/cloudfunctions.admin) - Application Integration Editor (
roles/integrations.integrationEditor) - Service Account User (
roles/iam.serviceAccountUser)
如需了解如何向主账号授予角色，请参阅授予、更改和撤消访问权限。
- Cloud Functions Admin (
- 如需连接到 Cloud Functions，请确保您已创建 OAuth 2.0 配置文件或已将用户管理的服务账号关联到您的集成：
-
如果您的集成关联了服务账号，请向该服务账号分配 Cloud Function Invoker IAM 角色。
如需了解如何向服务账号授予角色，请参阅管理对服务账号的访问权限。
- Cloud Functions 函数任务仅支持 Google OIDC ID 令牌类型的身份验证配置文件。使用分配有 Cloud Function Invoker IAM 角色的服务账号创建 Google OIDC ID 令牌类型的身份验证配置文件。如果 Cloud Functions 函数任务不需要身份验证，则任务配置窗格中的 Authentication profile 字段可以留空。
- 确保未在 Google Cloud 项目中为 Apigee Integration 设置 VPC Service Controls。
如果您的集成同时配置了 OIDC ID 配置文件和用户管理的服务账号，则默认情况下，系统会使用 OIDC ID 配置文件进行身份验证。如果 OIDC ID 配置文件和用户管理的服务账号均未配置，则系统会使用默认服务账号 (service-PROJECT_NUMBER@gcp-sa-apigee.iam.gserviceaccount.com) 调用 Cloud Functions 函数任务。
配置 Cloud Functions 函数任务
如需在集成中配置 Cloud Functions 函数任务，请执行以下步骤：
- 在 Apigee 界面中，选择您的 Apigee 组织。
- 点击开发 > 集成。
- 选择现有集成，或通过点击创建集成来创建新集成。
要创建新的集成，请执行以下操作：
- 在创建 Integrations对话框中输入名称和说明。
- 从支持的区域列表中选择集成的区域。
- 点击创建。
这将在集成设计器中打开集成。
- 在集成设计器导航栏中，点击 + 添加任务/触发器 > 任务以查看可用任务列表。
- 点击 Cloud Functions 函数元素，并将其放入集成编辑器中。
- 点击设计器上的 Cloud Functions 函数元素，以打开配置窗格，然后点击配置 Cloud Functions 函数 (Configure Cloud Function)。
-
在 Cloud Functions 函数配置 (Cloud Function Configuration) 对话框中，选择以下任一选项：
- 关联现有函数。选择此选项可关联在集成中配置的现有 Cloud Functions 函数。输入 Cloud Functions 函数触发器网址。
- 创建新函数 (Create new function)。选择此选项可创建与集成关联的新的 Cloud Functions 函数。输入 Cloud Functions 函数名称 (Cloud Function Name)，然后从下拉列表中选择函数区域 (function region)。
-
点击保存。
基本的 Google Cloud Functions 函数会在您的 Google Cloud 项目中创建，并与集成相关联。任务配置窗格会显示 Cloud Functions 函数的触发器网址和任务参数。
Cloud Functions 函数模板
使用现有 Cloud Functions 函数配置 Cloud Functions 函数时，请确保该函数的 main.py、task.py 和 requirements.txt 源文件采用以下格式：
task.py
# Sample Code:
# print(event.get('task_string_key'))
# event.set('task_int_array_key', [456, 789]);
# event.log('some logging')
def run(event):
"""Actual cloud function custom logic.
Args:
event : event object in main.py that contains all parameters.
"""
return
main.py
"""Un-editable platform wrapper which invokes user code."""
import traceback
from flask import json
from flask import jsonify
from task import run
VALUE_NAME = [
'stringValue', 'intValue', 'doubleValue', 'booleanValue', 'protoValue'
]
ARRAY_VALUE_NAME = {
'stringArray': 'stringValues',
'intArray': 'intValues',
'doubleArray': 'doubleValues',
'booleanArray': 'booleanValues',
'protoArray': 'protoValues'
}
VALUE_TYPE_URL = 'type.googleapis.com/google.protobuf.Value'
CLOUD_FUNCTION_EXCEPTION_KEY = 'CloudFunctionException'
CLOUD_FUNCTION_LOGGING_KEY = 'CloudFunctionLogging'
class _Event(object):
"""Event object."""
def __init__(self, json_payload):
self._event_params = json_payload.get('eventParameters', dict())
self._task_params = json_payload.get('taskParameters', dict())
self._log = []
print('Event param is ' + str(self._event_params))
print('Task param is ' + str(self._task_params))
def set(self, key, value):
"""Set the event parameters key-value.
Args:
key: parameter key.
value: parameter value.
"""
new_param = self._create_param(key, value)
param = self._get_param_by_key(key)
if param is None:
if 'parameters' not in self._event_params:
self._event_params['parameters'] = []
self._event_params['parameters'].append(new_param)
else:
param['value'] = new_param['value']
def _create_param(self, key, value):
"""Create a new parameter with given key value pair.
Args:
key: parameter key.
value: parameter value.
Returns:
parameter.
"""
new_param = {}
new_param['key'] = key
if isinstance(value, str):
new_param['value'] = {'stringValue': value}
elif isinstance(value, int):
new_param['value'] = {'intValue': value}
elif isinstance(value, float):
new_param['value'] = {'doubleValue': value}
elif isinstance(value, bool):
new_param['value'] = {'booleanValue': value}
elif isinstance(value, dict):
if 'type@' in value:
new_param['value'] = {'protoValue': value}
else:
new_param['value'] = {
'protoValue': {
'@type': 'type.googleapis.com/google.protobuf.Value',
'value': value
}
}
elif isinstance(value, list):
if not value:
raise RuntimeError('Cannot create a param with empty list')
if any(not isinstance(val, type(value[0])) for val in value):
print('Not all elements in the list have the same type')
new_param['value'] = {
'protoValue': {
'@type': 'type.googleapis.com/google.protobuf.Value',
'value': value
}
}
elif isinstance(value[0], str):
new_param['value'] = {'stringArray': {'stringValues': value}}
elif isinstance(value[0], int):
new_param['value'] = {'intArray': {'intValues': value}}
elif isinstance(value[0], float):
new_param['value'] = {'doubleArray': {'doubleValues': value}}
elif isinstance(value[0], bool):
new_param['value'] = {'booleanArray': {'booleanValues': value}}
elif isinstance(value[0], dict):
if all('@type' in val and val['@type'] == value[0]['@type']
for val in value):
new_param['value'] = {'protoArray': {'protoValues': value}}
else:
new_param['value'] = {
'protoValue': {
'@type': 'type.googleapis.com/google.protobuf.Value',
'value': value
}
}
else:
raise RuntimeError('The type ' + type(value[0]) +
' in the list is not supported')
else:
raise RuntimeError('Value ' + str(value) + ' has the type ' +
type(value) + ' that is not supported')
return new_param
def get(self, key):
"""Get the event parameter value for specified key.
Args:
key: parameter key.
Returns:
Parameter value.
"""
param = self._get_param_by_key(key)
if param is None:
raise RuntimeError('Can not find param with key ' + key)
return self._get_param_value(param)
def _get_param_by_key(self, key):
"""Get the parameter for specified key.
Args:
key: parameter key.
Returns:
Parameter.
"""
param = self._get_param_by_key_from_params(key, self._task_params)
if param is None:
return self._get_param_by_key_from_params(key, self._event_params)
value = self._get_param_value(param)
if isinstance(value, str) and len(value) > 2 and value.startswith(
'$') and value.endswith('$'):
return self._get_param_by_key_from_params(value[1:-1], self._event_params)
return param
def _get_param_by_key_from_params(self, key, params):
"""Get the parameter for specified key from event parameters.
Args:
key: parameter key.
params: event parameters.
Returns:
Parameter.
"""
if not isinstance(params, dict) or 'parameters' not in params:
return None
for param in params['parameters']:
if param['key'] == key:
return param
return None
def _get_param_value(self, param):
"""Get the parameter value for specified parameter.
Args:
param: parameter.
Returns:
Parameter value.
"""
value = param['value']
if len(value) != 1:
raise RuntimeError('param does not have size of 1')
for value_name in VALUE_NAME:
if value_name in value:
if value_name == 'protoValue' and value[value_name][
'@type'] == VALUE_TYPE_URL:
return value[value_name]['value']
return value[value_name]
for array_value_name in ARRAY_VALUE_NAME:
if array_value_name in value:
return value[array_value_name][ARRAY_VALUE_NAME[array_value_name]]
raise RuntimeError('Cannot get value from param ' + str(param))
def set_error(self):
"""Set the cloud function error to event parameters in order for user to see on IP."""
self.set(CLOUD_FUNCTION_EXCEPTION_KEY, traceback.format_exc())
def log(self, message):
self._log.append(str(message))
def get_response(self):
"""Get the response that can be returned to IP.
Returns:
The response text or any set of values that can be turned into a
Response object using
`make_response
<http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>`.
"""
if self._log:
self.set(CLOUD_FUNCTION_LOGGING_KEY, self._log)
res = {
'eventParameters': self._event_params,
}
return jsonify(**json.loads(json.htmlsafe_dumps(res)))
def execute_function(request):
"""Entry point of the cloud function.
Args:
request (flask.Request): HTTP request object.
Returns:
The response text or any set of values that can be turned into a
Response object using
`make_response
<http://flask.pocoo.org/docs/1.0/api/#flask.Flask.make_response>`.
"""
try:
request_json = request.get_json(silent=True)
event = _Event(request_json)
run(event)
except:
event.set_error()
return event.get_response()
requirements.txt
# Function dependencies, for example: # package>=version
修改 Cloud Functions 函数任务
在 Apigee Integration 中配置 Cloud Functions 函数任务会在您的 Google Cloud 项目中创建基本的 HTTP 触发的 Cloud Functions 函数。
如需修改 Cloud Functions 函数任务，请执行以下步骤：
- 在任务配置窗格中，点击打开 Cloud Functions 函数 (Open Cloud Function)。
系统会将您重定向到 Google Cloud 控制台中的函数详情页面。
- 点击修改。
- 您可以在配置页面中修改 Cloud Functions 函数的默认配置设置。如需了解详情，请参阅配置 Cloud Functions 函数。
- 点击下一步以修改 Cloud Functions 函数的源代码文件。
默认情况下，Cloud Functions 函数包含以下源文件：
- main.py：此文件包含用于在集成中运行 Cloud Functions 函数的初始化代码。
- task.py：此文件包含 Cloud Functions 函数的可执行代码。在
run(event)函数中编写脚本。此函数在 Cloud Functions 函数任务执行时调用。main.py 文件中的event对象包含所有任务参数。请参阅访问集成变量，了解如何在脚本中使用集成级别定义的变量。
- 点击部署。
访问集成变量
如需访问 Cloud Functions 函数中的集成变量，您必须以任务参数的形式将变量传递给 Cloud Functions 函数任务。任务参数是键值对，其中键是 Cloud Functions 函数源文件中使用的引用变量的名称，值是该引用变量指向的相应集成变量名称。您可以在任务配置窗格的任务参数部分添加一个或多个任务参数。
以下方法用于从 Cloud Functions 函数访问集成变量：
- set：将值写入变量。
- get：读取变量的值。
例如，如果您要在 Cloud Functions 函数源文件中使用名为 EmployeeName 的集成变量，请定义以下任务参数：
- 键：EmployeeKey
- 值：EmployeeName
以下示例脚本展示了如何使用 set 和 get 函数来访问定义的集成变量。
def run(event):
# Read the integration variable EmployeeName using the reference variable EmployeeKey
value = event.get('EmployeeKey');
# Change the integration variable EmployeeName value using the reference variable EmployeeKey
newValue = event.set('EmployeeKey' , 'XYZ');
# The new value of the integration variable is retained throughout the Cloud Function task.
return
错误处理策略
任务的错误处理策略指定当任务因暂时性错误而失败时要执行的操作。如需了解如何使用错误处理策略，以及了解不同类型的错误处理策略，请参阅错误处理策略。