有关使用 Workflows 进行无服务器编排的简介

使用 Workflows 简介无服务器编排

关于此 Codelab

上次更新时间：3月 19, 2025

atamel 编写

1. 简介

您可以使用 Workflows 创建无服务器工作流，以按照您定义的顺序将一系列无服务器任务关联起来。您可以将 Google Cloud API 和无服务器函数（如 Cloud Functions 和 Cloud Run）的强大功能结合使用，并调用外部 API，从而创建灵活的无服务器应用。

工作流不需要进行基础架构管理，并且可以无缝地按需扩缩，包括缩减至零。由于采用的是按用量付费价格模式，您只需为执行时间付费。

在本 Codelab 中，您将学习如何将各种 Google Cloud 服务和外部 HTTP API 与 Workflows 连接起来。具体而言，您将将两项公共 Cloud Functions 服务、一项专用 Cloud Run 服务和一个外部公共 HTTP API 连接到一个工作流中。

学习内容

• Workflows 基础知识。
• 如何将公共 Cloud Functions 函数与 Workflows 关联。
• 如何将专用 Cloud Run 服务与 Workflows 连接。
• 如何将外部 HTTP API 与 Workflows 相关联。

2. 设置和要求

自定进度的环境设置

1. 登录 Cloud 控制台，然后创建一个新项目或重复使用现有项目。 （如果您还没有 Gmail 或 G Suite 账号，则必须创建一个。）

注意：通过记住 Cloud Console 的网址 (console.cloud.google.com)，您可以轻松访问 Cloud Console。

请记住项目 ID，它在所有 Google Cloud 项目中都是唯一的名称（上述名称已被占用，您无法使用，抱歉！）。它稍后将在此 Codelab 中被称为 PROJECT_ID。

注意：如果您使用的是 Gmail 账号，可以将默认位置设置为无组织。如果您使用的是 G Suite 账号，请选择对贵组织来说合理的位置。

2. 接下来，您需要在 Cloud 控制台中启用结算功能，才能使用 Google Cloud 资源。

运行此 Codelab 应该不会产生太多的费用（如果有费用的话）。请务必按照“清理”部分中的所有说明操作，该部分介绍了如何关停资源，以免产生超出本教程范围的结算费用。Google Cloud 的新用户符合参与 $300 USD 免费试用计划的条件。

启动 Cloud Shell

虽然可以通过笔记本电脑对 Google Cloud 进行远程操作，但在此 Codelab 中，您将使用 Google Cloud Shell，这是一个在云端运行的命令行环境。

在 GCP 控制台中，点击右上角工具栏上的 Cloud Shell 图标：

预配和连接到环境应该只需要片刻时间。完成后，您应该会看到如下内容：

这个虚拟机已加载了您需要的所有开发工具。它提供了一个持久的 5GB 主目录，并且在 Google Cloud 中运行，大大增强了网络性能和身份验证功能。只需一个浏览器，即可完成本实验中的所有工作。

3. Workflows 概览

基础知识

工作流由一系列使用基于 YAML 的 Workflows 语法描述的步骤组成。这是工作流的定义。如需详细了解 Workflows YAML 语法，请参阅语法参考文档页面。

工作流创建完毕后即会进行部署，这样工作流便可以执行。执行是指单次运行工作流定义中包含的逻辑。所有工作流都会独立执行，并且该产品支持大量并发执行。

启用服务

在此 Codelab 中，您将将 Cloud Functions 函数和 Cloud Run 服务与 Workflows 相关联。您还将在构建服务期间使用 Cloud Build 和 Cloud Storage。

启用所有必要的服务：

gcloud services enable \
  cloudfunctions.googleapis.com \
  run.googleapis.com \
  workflows.googleapis.com \
  cloudbuild.googleapis.com \
  storage.googleapis.com

在下一步中，您将在工作流中将两个 Cloud Functions 函数关联起来。

4. 部署第一个 Cloud Functions 函数

第一个函数是 Python 中的随机数字生成器。

创建并前往函数代码所在的目录：

mkdir ~/randomgen
cd ~/randomgen

在该目录中创建一个 main.py 文件，其中包含以下内容：

import random, json
from flask import jsonify

def randomgen(request):
    randomNum = random.randint(1,100)
    output = {"random":randomNum}
    return jsonify(output)

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

该函数依赖于 Flask 进行 HTTP 处理，因此我们需要将其添加为依赖项。Python 依赖项可通过 pip 进行管理，并在名为 requirements.txt 的元数据文件中表示。

在同一目录中创建一个包含以下内容的 requirements.txt 文件：

flask>=1.0.2

使用以下命令部署具有 HTTP 触发器且允许未经身份验证的请求的函数：

gcloud functions deploy randomgen \
    --runtime python312 \
    --trigger-http \
    --allow-unauthenticated

函数部署后，您可以在控制台中或通过 gcloud functions describe 命令查看 url 属性下的函数网址。

您还可以使用以下 curl 命令访问该函数的网址：

curl $(gcloud functions describe randomgen --format='value(url)')

该函数已做好工作流准备。

5. 部署第二个 Cloud Functions 函数

第二个函数是乘数。它会将收到的输入值乘以 2。

创建并前往函数代码所在的目录：

mkdir ~/multiply
cd ~/multiply

在该目录中创建一个 main.py 文件，其中包含以下内容：

import random, json
from flask import jsonify

def multiply(request):
    request_json = request.get_json()
    output = {"multiplied":2*request_json['input']}
    return jsonify(output)

收到 HTTP 请求后，此函数会从 JSON 正文中提取 input，将其乘以 2，然后以 JSON 格式返回给调用方。

在同一目录中创建相同的 requirements.txt 文件，并在其中添加以下内容：

flask>=1.0.2

使用以下命令部署具有 HTTP 触发器且允许未经身份验证的请求的函数：

gcloud functions deploy multiply \
    --runtime python312 \
    --trigger-http \
    --allow-unauthenticated

函数部署后，您还可以使用以下 curl 命令访问该函数的网址：

curl $(gcloud functions describe multiply --format='value(url)') \
-X POST \
-H "content-type: application/json" \
-d '{"input": 5}'

该函数已做好工作流准备。

6. 连接两个 Cloud Functions 函数

在第一个工作流中，将这两个函数连接在一起。

创建一个包含以下内容的 workflow.yaml 文件。

请务必将网址值替换为包含正确区域和项目 ID 的函数实际网址

- randomgenFunction:
    call: http.get
    args:
        url: https://<region>-<project-id>.cloudfunctions.net/randomgen
    result: randomgenResult
- multiplyFunction:
    call: http.post
    args:
        url: https://<region>-<project-id>.cloudfunctions.net/multiply
        body:
            input: ${randomgenResult.body.random}
    result: multiplyResult
- returnResult:
    return: ${multiplyResult}

在此工作流中，您从第一个函数获取一个随机数字，并将其传递给第二个函数。结果就是乘积的随机数。

部署第一个工作流：

gcloud workflows deploy workflow --source=workflow.yaml

执行第一个工作流：

gcloud workflows execute workflow

工作流执行完毕后，您可以通过传入上一步中提供的执行 ID 来查看结果：

gcloud workflows executions describe <your-execution-id> --workflow workflow

输出将包含 result 和 state：

result: '{"body":{"multiplied":108},"code":200 ... } 

...
state: SUCCEEDED

7. 连接外部 HTTP API

接下来，您将在工作流中将 math.js 连接为外部服务。

在 math.js 中，您可以按如下方式求解数学表达式：

curl https://api.mathjs.org/v4/?'expr=log(56)'

这次，您将使用 Cloud 控制台更新我们的工作流。在 Google Cloud 控制台中找到 Workflows：

找到您的工作流，然后点击 Definition 标签页：

修改工作流定义，并添加对 math.js 的调用。

请务必将网址值替换为包含正确区域和项目 ID 的函数实际网址

- randomgenFunction:
    call: http.get
    args:
        url: https://<region>-<project-id>.cloudfunctions.net/randomgen
    result: randomgenResult
- multiplyFunction:
    call: http.post
    args:
        url: https://<region>-<project-id>.cloudfunctions.net/multiply
        body:
            input: ${randomgenResult.body.random}
    result: multiplyResult
- logFunction:
    call: http.get
    args:
        url: https://api.mathjs.org/v4/
        query:
            expr: ${"log(" + string(multiplyResult.body.multiplied) + ")"}
    result: logResult
- returnResult:
    return: ${logResult}

工作流现在会将乘法函数的输出馈送到 math.js 中的日志函数调用。

界面会引导您修改和部署工作流。部署完成后，点击 Execute 以执行工作流。您会看到执行的详细信息：

请注意日志函数的输出中包含状态代码 200 和 body。

您刚刚将外部服务集成到我们的工作流中，太棒了！

8. 部署 Cloud Run 服务

在最后一部分，通过调用专用 Cloud Run 服务来完成工作流。这意味着，工作流需要经过身份验证才能调用 Cloud Run 服务。

Cloud Run 服务会返回传入数字的 math.floor。

创建并前往服务代码目录：

mkdir ~/floor
cd ~/floor

在该目录中创建一个 app.py 文件，其中包含以下内容：

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)))

Cloud Run 会部署容器，因此您需要 Dockerfile，并且您的容器需要绑定到 0.0.0.0 和 PORT 环境变量，因此需要使用上述代码。

收到 HTTP 请求后，此函数会从 JSON 正文中提取 input，调用 math.floor 并将结果返回给调用方。

在同一目录中，创建以下 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

构建容器。

export SERVICE_NAME=floor
gcloud builds submit --tag gcr.io/${GOOGLE_CLOUD_PROJECT}/${SERVICE_NAME}

容器构建完成后，将其部署到 Cloud Run。请注意 no-allow-unauthenticated 标志。这样可确保服务仅接受经过身份验证的调用：

gcloud run deploy ${SERVICE_NAME} \
  --image gcr.io/${GOOGLE_CLOUD_PROJECT}/${SERVICE_NAME} \
  --platform managed \
  --no-allow-unauthenticated

部署完成后，服务便可用于工作流。

9. 连接 Cloud Run 服务

在配置 Workflows 以调用专用 Cloud Run 服务之前，您需要创建一个供 Workflows 使用的服务账号：

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

向服务账号授予 run.invoker 角色。这样，服务账号便可以调用经过身份验证的 Cloud Run 服务：

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

更新 workflow.yaml 中的工作流定义，以添加 Cloud Run 服务。请注意，您还需要添加 auth 字段，以确保 Workflows 在调用 Cloud Run 服务时传递身份验证令牌：

请务必将网址值替换为包含正确区域和项目 ID 的函数实际网址

- randomgenFunction:
    call: http.get
    args:
        url: https://<region>-<project-id>.cloudfunctions.net/randomgen
    result: randomgenResult
- multiplyFunction:
    call: http.post
    args:
        url: https://<region>-<project-id>.cloudfunctions.net/multiply
        body:
            input: ${randomgenResult.body.random}
    result: multiplyResult
- logFunction:
    call: http.get
    args:
        url: https://api.mathjs.org/v4/
        query:
            expr: ${"log(" + string(multiplyResult.body.multiplied) + ")"}
    result: logResult
- floorFunction:
    call: http.post
    args:
        url: https://floor-<random-hash>.run.app
        auth:
            type: OIDC
        body:
            input: ${logResult.body}
    result: floorResult
- returnResult:
    return: ${floorResult}

更新工作流。这次传入服务账号：

gcloud workflows deploy workflow \
    --source=workflow.yaml \
    --service-account=${SERVICE_ACCOUNT}@${GOOGLE_CLOUD_PROJECT}.iam.gserviceaccount.com

执行工作流：

gcloud workflows execute workflow

几秒钟后，您可以查看工作流执行情况，了解结果：

gcloud workflows executions describe <your-execution-id> --workflow workflow

输出将包含整数 result 和 state：

result: '{"body":"5","code":200 ... } 

...
state: SUCCEEDED

10. 恭喜！

恭喜您完成此 Codelab。

所学内容

• Workflows 基础知识。
• 如何将公共 Cloud Functions 函数与 Workflows 关联。
• 如何将专用 Cloud Run 服务与 Workflows 连接。
• 如何将外部 HTTP API 与 Workflows 相关联。