Cloud Functions 函数任务

借助 Cloud Functions 函数任务，您可以从集成中配置和运行 Google Cloud Functions。Google Cloud Functions 是一个轻量级计算解决方案，可供开发者创建单一用途的独立函数，无需管理服务器或运行时环境即可对 Cloud 事件作出响应。

如需了解详情，请参阅 Google Cloud Functions 文档。

须知事项

在配置 Cloud Functions 函数任务之前，请确保在您的 Google Cloud 项目中执行以下任务。

启用 Cloud Functions API (cloudfunctions.googleapis.com)。
启用 Cloud Functions API
将以下 IAM 角色分配给您的主账号：
- Cloud Functions Admin (roles/cloudfunctions.admin)
- Application Integration Editor (roles/integrations.integrationEditor)
- Service Account User (roles/iam.serviceAccountUser)
如需了解如何向主账号授予角色，请参阅授予、更改和撤消访问权限。
如需连接到 Cloud Run 函数，请确保您已创建 OAuth 2.0 配置文件或为用户代管式服务账号关联到您的集成：

如果您的集成关联了服务账号，请向该服务账号分配 Cloud Function Invoker IAM 角色。
如需了解如何向服务账号授予角色，请参阅管理对服务账号的访问权限。
Cloud Functions 函数任务仅支持 Google OIDC ID 令牌类型的身份验证配置文件。使用分配有 Cloud Function Invoker IAM 角色的服务账号创建 Google OIDC ID 令牌类型的身份验证配置文件。如果 Cloud Functions 函数任务不需要身份验证，则任务配置窗格中的 Authentication profile 字段可以留空。

如果您的集成同时配置了 OIDC ID 配置文件和用户管理的服务账号，则默认情况下，系统会使用 OIDC ID 配置文件进行身份验证。如果 OIDC ID 配置文件和用户管理的服务账号均未配置，则系统会使用默认服务账号 (service-PROJECT_NUMBER@gcp-sa-integrations.iam.gserviceaccount.com) 调用 Cloud Functions 函数任务。

确保未在 Google Cloud 项目中为应用集成设置 VPC Service Controls。
警告：如果您在 Google Cloud 项目中为 Application Integration 设置了 VPC Service Controls，则 Cloud Functions 无法运行或停止运行。

配置 Cloud Functions 函数任务

如需在集成中配置 Cloud Functions 函数任务，请执行以下步骤：

在 Google Cloud 控制台中，前往 Application Integration 页面。
转到 Application Integration
在导航菜单中，点击集成。
系统会显示集成页面，其中列出了 Google Cloud 项目中可用的所有集成。
选择现有集成，或点击创建集成以创建新的集成。
要创建新的集成，请执行以下操作：
1. 在创建集成窗格中输入名称和说明。
2. 选择集成的区域。
  注意：区域下拉列表仅列出您的 Google Cloud 项目中预配的区域。如需预配新区域，请点击 Enable Region。如需了解详情，请参阅启用新区域。
3. 为集成选择服务账号。您随时可以在集成工具栏的 集成摘要窗格中更改或更新集成的服务账号详细信息。
  注意：只有当您为所选区域启用了集成治理功能时，系统才会显示用于选择服务账号的选项。
4. 点击创建。
系统随即会在集成编辑器中打开该集成。
在集成编辑器导航栏中，点击任务查看可用任务和连接器的列表。
点击 Cloud Functions 函数元素，并将其放入集成编辑器中。
点击设计器上的 Cloud Functions 函数元素，以打开配置窗格，然后点击配置 Cloud Functions 函数 (Configure Cloud Function)。
在 Cloud Functions 函数配置 (Cloud Function Configuration) 对话框中，选择以下任一选项：
- 关联现有函数。选择此选项可关联在集成中配置的现有 Cloud Functions 函数。输入 Cloud Functions 函数触发器网址。
  重要提示：请确保现有 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 函数任务

在 Application Integration 中配置 Cloud Functions 函数任务可创建基本的 HTTP 在您的 Google Cloud 项目中触发了 Cloud Function。

如需修改 Cloud Functions 函数任务，请执行以下步骤：

在任务配置窗格中，点击打开 Cloud Functions 函数 (Open Cloud Function)。
系统会将您重定向到 Google Cloud 控制台中的函数详情页面。
点击修改。
您可以在配置页面中修改 Cloud Functions 函数的默认配置设置。如需了解详情，请参阅配置 Cloud Functions 函数。
点击下一步以修改 Cloud Functions 函数的源代码文件。
默认情况下，Cloud Functions 函数包含以下源文件：
- main.py：此文件包含用于在集成中运行 Cloud Functions 函数的初始化代码。
  注意：请勿修改 main.py 文件。
- task.py：此文件包含 Cloud Functions 函数的可执行代码。在 run(event) 函数中编写脚本。此函数在 Cloud Functions 函数任务执行时调用。main.py 文件中的 event 对象包含所有任务参数。
  请参阅访问集成变量，了解如何在脚本中使用集成级别定义的变量。
点击部署。

访问集成变量

如需访问 Cloud Functions 函数中的集成变量，您必须以任务参数的形式将变量传递给 Cloud Functions 函数任务。任务参数是键值对，其中“键”是指您的 Cloud Function 函数源文件中使用的引用变量的名称，“值”是引用变量指向的对应集成变量名称。您可以在任务配置窗格的任务参数部分添加一个或多个任务参数。

以下方法用于从 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

错误处理策略

任务的错误处理策略指定当任务因暂时性错误而失败时要执行的操作。如需了解如何使用错误处理策略，以及了解不同类型的错误处理策略，请参阅错误处理策略。

SLA 例外情况

Cloud Functions 函数任务依赖于 Google Cloud Functions 产品。由于此依赖项位于应用集成之外，因此因 Cloud Functions 任务失败而导致的所有 active 集成执行失败情况均不受应用集成服务等级协议 (SLA) 条款及条件的约束。

配额和限制

如需了解配额和限制，请参阅配额和限制。

后续步骤

了解所有任务和触发器。
了解如何测试和发布集成。

了解错误处理。
了解集成执行日志。