Cloud Function 任务

借助 Cloud Functions 函数任务，您可以从集成中配置和运行 Cloud Run functions。Cloud Run functions 代表 Cloud Functions 的下一代产品，它提供事件驱动型编程模型，具有增强的控制和可伸缩性，并利用 Cloud Run 无服务器平台。Cloud Run functions 为所有工作负载类型提供单一的无服务器解决方案。

Cloud Functions 函数任务支持以下版本的 Cloud Run functions：

如需详细了解 Cloud Run functions 各版本之间的差异，请参阅 Cloud Functions 比较 指南。

准备工作

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

  1. 如需连接到 Cloud Function，请确保您已创建 OAuth 2.0 配置文件或已将用户代管式服务账号关联到您的集成：
    • 如果您的集成关联了服务帐号，请向该服务帐号分配 Cloud Functions Invoker IAM 角色。

      如需了解如何向服务帐号授予角色，请参阅管理对服务账号的访问权限。

    • Cloud Functions 函数任务仅支持 Google OIDC ID 令牌 类型的身份验证配置文件。 使用分配有 Cloud Functions Invoker IAM 角色的服务帐号创建 Google OIDC ID 令牌类型的身份验证配置文件。 如果 Cloud Functions 函数任务不需要身份验证， 则任务配置窗格中的 身份验证配置文件 字段可以留空。

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

  2. 确保未在您的 Google Cloud 项目中为 Application Integration 设置 VPC Service Controls。

配置 Cloud Function 任务

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

  1. 在导航菜单中，点击集成 。

    这会打开 Application Integration 界面中可用集成的列表。

  2. 选择现有集成或点击创建集成 。

    如果您要创建新集成，请在 创建对话框中输入名称和说明，然后点击 创建。

  3. 从任务 下拉列表中，点击 Cloud Functions ，将其放置在 集成编辑器中。
  4. 点击设计器上的 Cloud Functions 函数元素，以 打开配置窗格，然后点击 配置 Cloud Functions 函数。
  5. 如果系统提示您向服务帐号授予权限，请点击 授予。

    Application Integration 会自动向 服务帐号授予所需的权限。

  6. 在 Cloud Functions 配置 窗格中，选择以下选项之一：
    • 关联现有函数 ：选择此选项可将现有函数与您的集成相关联。您可以从集成中关联 Cloud Functions（第 1 代）和使用 Cloud Functions v2 API 创建的 Cloud Functions。
      • 在 Cloud Functions 函数触发器网址 字段中，输入现有函数的触发器网址。

        网址应采用以下格式之一：

        # For Cloud Functions (1st gen)
https://REGION_NAME-PROJECT_ID.cloudfunctions.net/FUNCTION_NAME
         
        # For Cloud Run functions created using the Cloud Functions v2 API
https://FUNCTION_NAME-PROJECT_ID.REGION_NAME.run.app

    • 创建新函数 ：选择此选项可为集成创建新函数。
      1. 在函数名称 字段中，为新的 Cloud Run functions 输入一个唯一名称。
      2. 从区域 下拉列表中，选择 Cloud Run functions 的部署区域。
      3. 从函数版本 下拉列表中，选择所需的 Cloud Run functions 版本：
        • Cloud Functions（第 1 代） ：这是 Cloud Run functions 的旧版，以前称为 Cloud Functions（第 1 代），它使用 .cloudfunctions.net 端点格式。
        • Cloud Functions（最新一代） ：这是使用 Cloud Functions v2 API 创建的最新版 Cloud Run functions。它基于 Cloud Run 和 Eventarc 构建，支持延长请求超时时间（最长 60 分钟），提供更高的并发，并使用 .cloudfunctions.net 和 .run.app 端点格式。
          • 如需详细了解这两个版本之间的差异，请参阅比较 Cloud Functions。
  7. 点击保存 。

    8. 在 Application Integration 中配置 Cloud Functions 函数 任务会在您的 Google Cloud 云项目 中创建基本的 HTTP- 触发的 Cloud Run functions。

Cloud Functions 函数模板

以下示例展示了如何在集成中使用 Cloud Functions 函数任务（针对不同语言）。

Python

使用现有 Cloud Run functions 配置 Cloud Function 时，请确保该函数的 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'