教程：将 Workflows 与 Cloud Run 和 Cloud Run 函数搭配使用

部署第一项 Cloud Run 函数服务

收到 HTTP 请求后，此 HTTP 函数会生成一个介于 1 到 100 之间的随机数字，然后以 JSON 格式返回该数字。

1. 创建名为 randomgen 的目录并切换到该目录：

   mkdir ~/randomgen
   cd ~/randomgen

2. 创建一个文件名为 main.py 且包含以下 Python 代码的文本文件：

   import functions_framework
   import random
   from flask import jsonify
   
   
   @functions_framework.http
   def randomgen(request):
       randomNum = random.randint(1, 100)
       output = {"random": randomNum}
       return jsonify(output)

3. 如需支持依赖于 Flask 进行 HTTP 处理，请为 pip 软件包管理器创建一个文本文件。为该文本文件指定文件名 requirements.txt 并添加以下内容：

   flask>=1.0.2
   functions-framework==3.0.0

4. 为 Workflows 创建服务账号，以供使用：

   export SERVICE_ACCOUNT=workflows-sa
   gcloud iam service-accounts create ${SERVICE_ACCOUNT}

5. 如需允许服务账号调用经过身份验证的 Cloud Run 服务，请向 Workflows 服务账号授予 run.invoker 角色：

   gcloud projects add-iam-policy-binding PROJECT_ID \
       --member "serviceAccount:${SERVICE_ACCOUNT}@PROJECT_ID.iam.gserviceaccount.com" \
       --role "roles/run.invoker"

6. 使用 HTTP 触发器部署函数，并允许未经身份验证的访问：

   gcloud functions deploy randomgen-function \
       --gen2 \
       --runtime python310 \
       --entry-point=randomgen \
       --trigger-http \
       --allow-unauthenticated

   部署该函数可能需要几分钟的时间。或者，您也可以使用 Google Cloud 控制台中的 Cloud Run 函数界面来部署函数。

7. 部署 randomgen 函数后，您可以确认 httpsTrigger.url 属性：

   gcloud functions describe randomgen-function \
       --gen2 \
       --format="value(serviceConfig.uri)"

8. 保存网址。 在后续练习中，您需要将它添加到工作流源文件中。

9. 您可以通过以下 curl 命令来试用该函数：

   curl $(gcloud functions describe randomgen-function \
       --gen2 \
       --format="value(serviceConfig.uri)")

   系统随机会生成一个数字并返回。

部署第二项 Cloud Run 函数服务

收到 HTTP 请求后，此 HTTP 函数会从 JSON 正文中提取 input，将此数字乘以 2，然后以 JSON 格式返回结果。

1. 返回到您的主目录：

   cd ~

2. 创建名为 multiply 的目录并切换到该目录：

   mkdir ~/multiply
   cd ~/multiply

3. 创建一个文件名为 main.py 且包含以下 Python 代码的文本文件：

   import functions_framework
   from flask import jsonify
   
   
   @functions_framework.http
   def multiply(request):
       request_json = request.get_json()
       output = {"multiplied": 2 * request_json['input']}
       return jsonify(output)

4. 如需支持依赖于 Flask 进行 HTTP 处理，请为 pip 软件包管理器创建一个文本文件。为该文本文件指定文件名 requirements.txt 并添加以下内容：

   flask>=1.0.2
   functions-framework==3.0.0

5. 使用 HTTP 触发器部署函数，并允许未经身份验证的访问：

   gcloud functions deploy multiply-function \
       --gen2 \
       --runtime python310 \
       --entry-point=multiply \
       --trigger-http \
       --allow-unauthenticated

   部署函数可能需要几分钟的时间。或者，您也可以使用 Google Cloud 控制台中的 Cloud Run 函数界面来部署函数。

6. 部署 multiply 函数后，您可以确认 httpsTrigger.url 属性：

   gcloud functions describe multiply-function \
       --gen2\
       --format="value(serviceConfig.uri)"

7. 保存网址。 在后续练习中，您需要将它添加到工作流源文件中。

8. 您可以通过以下 curl 命令来试用该函数：

   curl -X POST MULTIPLY_FUNCTION_URL \
       -H "Authorization: Bearer $(gcloud auth print-identity-token)" \
       -H "Content-Type: application/json" \
       -d '{"input": 5}'

   系统应会返回数字 10。

在工作流中连接两项 Cloud Run 函数服务

工作流由一系列使用 Workflows 语法描述的步骤组成，该语法可以采用 YAML 或 JSON 格式编写。这是工作流的定义。如需了解详细说明，请参阅语法参考文档页面。

1. 返回到您的主目录：

   cd ~

2. 创建一个文件名为 workflow.yaml 且包含以下内容的文本文件：

   - randomgen_function:
       call: http.get
       args:
           url: RANDOMGEN_FUNCTION_URL
       result: randomgen_result
   - multiply_function:
       call: http.post
       args:
           url: MULTIPLY_FUNCTION_URL
           body:
               input: ${randomgen_result.body.random}
       result: multiply_result
   - return_result:
       return: ${multiply_result}
   

   • 将 RANDOMGEN_FUNCTION_URL 替换为您的 randomgen 函数的网址。
   • 将 MULTIPLY_FUNCTION_URL 替换为您的 multiply 函数的网址。

   此源文件会将两个 HTTP 函数关联在一起，并返回最终结果。

3. 创建工作流后，可以进行部署，使其可以执行。

   gcloud workflows deploy WORKFLOW_NAME \
       --source=workflow.yaml

   将 WORKFLOW_NAME 替换为您的工作流的名称。

4. 执行工作流：

   gcloud workflows run WORKFLOW_NAME

   执行是指单次运行工作流定义中包含的逻辑。所有工作流都会独立执行，并且 Workflows 的快速扩缩允许大量并发执行。

   执行工作流后，输出应类似于以下内容：

   result: '{"body":{"multiplied":120},"code":200,"headers":{"Alt-Svc":"h3-29=\":443\";
   ...
   startTime: '2021-05-05T14:17:39.135251700Z'
   state: SUCCEEDED
   ...
   

在工作流中连接公共 REST 服务

更新现有工作流并连接可以用来对数学表达式求值的公共 REST API (math.js)。例如 curl https://api.mathjs.org/v4/?'expr=log(56)'。

请注意，您已经部署了工作流，现在您可以通过 Google Cloud 控制台中的“工作流”页面对其进行修改。

1. 修改工作流的源文件并将其替换为以下内容：

   - randomgen_function:
       call: http.get
       args:
           url: RANDOMGEN_FUNCTION_URL
       result: randomgen_result
   - multiply_function:
       call: http.post
       args:
           url: MULTIPLY_FUNCTION_URL
           body:
               input: ${randomgen_result.body.random}
       result: multiply_result
   - log_function:
       call: http.get
       args:
           url: https://api.mathjs.org/v4/
           query:
               expr: ${"log(" + string(multiply_result.body.multiplied) + ")"}
       result: log_result
   - return_result:
       return: ${log_result}
   

   • 将 RANDOMGEN_FUNCTION_URL 替换为您的 randomgen 函数的网址。
   • 将 MULTIPLY_FUNCTION_URL 替换为您的 multiply 函数的网址。

   这会将外部 REST 服务与 Cloud Run 函数服务相关联，并返回最终结果。

2. 部署修改后的工作流：

   gcloud workflows deploy WORKFLOW_NAME \
       --source=workflow.yaml

部署 Cloud Run 服务

部署 Cloud Run 服务，在收到 HTTP 请求后，该服务会从 JSON 正文中提取 input，计算其 math.floor，并返回结果。

1. 创建名为 floor 的目录并切换到该目录：

   mkdir ~/floor
   cd ~/floor

2. 创建一个文件名为 app.py 且包含以下 Python 代码的文本文件：

   import json
   import logging
   import os
   import math
   
   from flask import Flask, request
   
   app = Flask(__name__)
   
   
   @app.route('/', methods=['POST'])
   def handle_post():
       content = json.loads(request.data)
       input = float(content['input'])
       return f"{math.floor(input)}", 200
   
   
   if __name__ != '__main__':
       # Redirect Flask logs to Gunicorn logs
       gunicorn_logger = logging.getLogger('gunicorn.error')
       app.logger.handlers = gunicorn_logger.handlers
       app.logger.setLevel(gunicorn_logger.level)
       app.logger.info('Service started...')
   else:
       app.run(debug=True, host='0.0.0.0', port=int(os.environ.get('PORT', 8080)))

3. 在同一目录中，创建一个包含以下内容的 Dockerfile：

   # Use an official lightweight Python image.
   # https://hub.docker.com/_/python
   FROM python:3.7-slim
   
   # Install production dependencies.
   RUN pip install Flask gunicorn
   
   # Copy local code to the container image.
   WORKDIR /app
   COPY . .
   
   # Run the web service on container startup. Here we use the gunicorn
   # webserver, with one worker process and 8 threads.
   # For environments with multiple CPU cores, increase the number of workers
   # to be equal to the cores available.
   CMD exec gunicorn --bind 0.0.0.0:8080 --workers 1 --threads 8 app:app

4. 创建一个 Artifact Registry 标准制品库，您可以在其中存储您的 Docker 容器映像：

   gcloud artifacts repositories create REPOSITORY \
       --repository-format=docker \
       --location=${REGION}

   将 REPOSITORY 替换为制品库的唯一名称。

5. 构建容器映像：

   export SERVICE_NAME=floor
   gcloud builds submit --tag ${REGION}-docker.pkg.dev/PROJECT_ID/REPOSITORY/${SERVICE_NAME}

6. 将容器映像部署到 Cloud Run，确保其仅接受经过身份验证的调用：

   gcloud run deploy ${SERVICE_NAME} \
       --image ${REGION}-docker.pkg.dev/PROJECT_ID/REPOSITORY/${SERVICE_NAME}:latest \
       --no-allow-unauthenticated

当您看到服务网址时，表示部署完成。 在更新工作流定义时，您需要指定该网址。

在工作流中连接 Cloud Run 服务

更新现有工作流并指定 Cloud Run 服务的网址。

1. 修改工作流的源文件并将其替换为以下内容：

   - randomgen_function:
       call: http.get
       args:
           url: RANDOMGEN_FUNCTION_URL
       result: randomgen_result
   - multiply_function:
       call: http.post
       args:
           url: MULTIPLY_FUNCTION_URL
           body:
               input: ${randomgen_result.body.random}
       result: multiply_result
   - log_function:
       call: http.get
       args:
           url: https://api.mathjs.org/v4/
           query:
               expr: ${"log(" + string(multiply_result.body.multiplied) + ")"}
       result: log_result
   - floor_function:
       call: http.post
       args:
           url: CLOUD_RUN_SERVICE_URL
           auth:
               type: OIDC
           body:
               input: ${log_result.body}
       result: floor_result
   - create_output_map:
       assign:
         - outputMap:
             randomResult: ${randomgen_result}
             multiplyResult: ${multiply_result}
             logResult: ${log_result}
             floorResult: ${floor_result}
   - return_output:
       return: ${outputMap}
   

   • 将 RANDOMGEN_FUNCTION_URL 替换为您的 randomgen 函数的网址。
   • 将 MULTIPLY_FUNCTION_URL 替换为您的 multiply 函数的网址。
   • 将 CLOUD_RUN_SERVICE_URL 替换为您的 Cloud Run 服务网址。

   这将连接工作流中的 Cloud Run 服务。请注意，auth 密钥可确保在调用 Cloud Run 服务时传递身份验证令牌。如需了解详情，请参阅通过工作流发出经过身份验证的请求。

2. 部署修改后的工作流：

   gcloud workflows deploy WORKFLOW_NAME \
       --source=workflow.yaml

3. 执行最终工作流：

   gcloud workflows run WORKFLOW_NAME

   您应该会看到类似如下所示的输出：

   result: '{"Floor":{"body":"4","code":200
     ...
     "Log":{"body":"4.02535169073515","code":200
     ...
     "Multiply":{"body":{"multiplied":56},"code":200
     ...
     "Random":{"body":{"random":28},"code":200
     ...
   startTime: '2023-11-13T21:22:56.782669001Z'
   state: SUCCEEDED
   

恭喜！您已部署并执行一个将一系列服务连接在一起的工作流。

如需使用表达式、条件跳转、Base64 编码或解码、子工作流等创建更复杂的工作流，请参阅工作流语法参考文档和标准库概览。