Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

Cloud Function 任务

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

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

Cloud Functions（第 1 代）
使用 Cloud Functions v2 API 创建的 Cloud Run functions

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

准备工作

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

如需连接到 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.) 调用 Cloud Functions 函数任务。

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

配置 Cloud Function 任务

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

在导航菜单中，点击集成。
这会打开 Application Integration 界面中可用集成的列表。
选择现有集成或点击创建集成 。
如果您要创建新集成，请在创建对话框中输入名称和说明，然后点击创建。
从任务下拉列表中，点击 Cloud Functions ，将其放置在集成编辑器中。
点击设计器上的 Cloud Functions 函数元素，以打开配置窗格，然后点击 配置 Cloud Functions 函数。
如果系统提示您向服务帐号授予权限，请点击授予。
Application Integration 会自动向服务帐号授予所需的权限。
在 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
```
  重要提示：确保现有 Cloud Function 的源文件符合 Cloud Functions 函数模板中指定的格式。
- 创建新函数 ：选择此选项可为集成创建新函数。
  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 端点格式。
点击保存。

在 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'] = {
                '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(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

如需详细了解响应格式，请参阅 ValueType。

Java

以下示例展示了如何在集成中使用 Cloud Functions 函数任务。确保响应遵循示例中所述的受支持 JSON 格式：

private static final Gson gson = new Gson();

@Override
public void service(HttpRequest request, HttpResponse response) throws Exception {
  JsonObject body = gson.fromJson(request.getReader(), JsonObject.class);

  JsonArray resParams = new JsonArray();
  for (JsonElement param: body.getAsJsonObject("eventParameters").getAsJsonArray("parameters")) {
    if (param.getAsJsonObject().get("key").getAsString().equals("input")) {
      JsonObject newParam= new JsonObject();
      newParam.addProperty("key", "input");
      JsonObject value = new JsonObject();
      value.addProperty("stringValue","2");
      newParam.add("value", value);
      resParams.add(newParam);
    } else {
      resParams.add(param);
    }
  }
  JsonObject parameters = new JsonObject();
  parameters.add("parameters", resParams);
  JsonObject res = new JsonObject();
  res.add("eventParameters", parameters);
  System.out.println(res);
  BufferedWriter writer = response.getWriter();
  writer.write(res.toString());
}

如需详细了解响应格式，请参阅 ValueType。

JavaScript

以下示例展示了如何在集成中使用 Cloud Functions 函数任务。确保响应遵循示例中所述的受支持 JSON 格式：

const functions = require('@google-cloud/functions-framework');

functions.http('execute_function', (req, res) => {
  console.log(JSON.stringify(req.body));
  let response = {"eventParameters":{"parameters":[{"key":"input","value":{"stringValue":"2"}}]}};
  res.send(JSON.stringify(response));
});

如需详细了解响应格式，请参阅 ValueType。

PHP

以下示例展示了如何在集成中使用 Cloud Functions 函数任务。确保响应遵循示例中所述的受支持 JSON 格式：

use Psr\Http\Message\ServerRequestInterface;
function execute_function(ServerRequestInterface $request)
{
  return '{"eventParameters":{"parameters":[{"key":"input","value":{"stringValue":"2"}}]}}';
}

如需详细了解响应格式，请参阅 ValueType。

修改 Cloud Function 任务

Application Integration 会根据版本类型将您定向到相应的 Google Cloud 控制台页面，以修改 Cloud Run functions。

Cloud Functions（第 1 代）

如需修改使用 Cloud Functions（第 1 代）版本配置的 Cloud Function 任务，请执行以下步骤：

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

Cloud Run functions

如需修改使用 Cloud Functions（最新一代）版本配置的 Cloud Function 任务，请执行以下步骤：

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

访问集成变量

如需访问 Cloud Function 中的集成变量，请将该变量作为任务参数传递给 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
  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 产品。由于此依赖项位于 Application Integration 外部，因此，由于 Cloud Function 任务失败而导致的所有 active 集成执行失败均不属于 Application Integration 服务等级协议 (SLA) 条款和条件的范围。

配额和限制

Cloud Functions 函数任务的执行超时时间限制为 2 分钟（120 秒） 。如果调用的函数需要超过 2 分钟才能完成并响应，则任务会因执行超时错误 (TIMEOUT_WEB) 而失败。

注意：即使底层 Cloud Run functions 配置了更长的请求超时时间（例如，最长 60 分钟），或者您的集成采用异步 (ASYNC) 执行策略运行，Cloud Functions 函数也会同步等待 HTTP 响应，并在 2 分钟后超时。

如果您的工作负载需要超过 2 分钟的执行时间，请考虑以下解决方法：

使用 Pub/Sub 的异步触发器 ：在集成中使用 Pub/Sub 任务或连接器发布事件消息，并将 Cloud Run functions 配置为从 Pub/Sub 主题订阅异步触发，以便在后台处理工作负载。
即时 HTTP 确认和异步回调 ：将 HTTP 触发的 Cloud Run functions 配置为在收到请求后立即返回 HTTP 200 OK 响应，在后台异步处理长时间运行的任务，并可以选择调用 Application Integration API 触发器端点，以便在处理完成后恢复下游集成步骤。

如需了解 Cloud Run functions 和 Cloud Functions（第 1 代）的一般配额和限制，请参阅配置比较。

后续步骤

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

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