如何通过原生方法验证Google Cloud Python SDK防火墙部署是否成功

有官方原生方法可以直接校验部署状态，不需要自行拉取防火墙规则列表做人工比对。

实现原理 #

GCP 所有资源的增删改操作都会返回对应的 Operation 资源，该资源会全程跟踪操作的执行状态，你可以直接通过该资源获取部署的最终结果。VPC防火墙属于全局资源，对应的操作可以通过全局Operation接口查询状态。

推荐实现方案 #

方案1：阻塞等待部署完成（最常用） #

调用GlobalOperationsClient的wait()方法，会自动阻塞到部署操作完成，直接返回最终执行结果：

import time
from google.cloud import compute_v1

# 初始化客户端
firewall_client = compute_v1.FirewallsClient()
operation_client = compute_v1.GlobalOperationsClient()

# 构造你的防火墙规则，此处省略具体参数构造逻辑
firewall_rule = compute_v1.Firewall(
    name="test-firewall",
    network="projects/你的项目ID/global/networks/你的VPC名称",
    allowed=[compute_v1.Allowed(I_p_protocol="tcp", ports=["80"])]
)

# 发起防火墙创建请求，拿到Operation对象
operation = firewall_client.insert(
    project="你的项目ID",
    firewall_resource=firewall_rule
)

# 原生阻塞等待部署完成
operation_result = operation_client.wait(
    project="你的项目ID",
    operation=operation.name
)

# 校验部署结果
if operation_result.error:
    print(f"部署失败，错误详情：{operation_result.error}")
else:
    print("部署完全成功，防火墙规则已全局生效")

方案2：非阻塞轮询状态 #

如果不想阻塞主线程，可以自行轮询Operation状态判断部署进度：

while True:
    current_op = operation_client.get(
        project="你的项目ID",
        operation=operation.name
    )
    # 操作执行完成
    if current_op.status == compute_v1.Operation.Status.DONE:
        if current_op.error:
            print(f"部署失败：{current_op.error}")
        else:
            print("部署成功")
        break
    # 间隔2秒轮询一次
    time.sleep(2)

方案优势 #

• 无需自行实现规则字段比对逻辑，API会自动确认操作的最终执行结果
• 可以捕获部署过程中的所有错误，包括权限不足、资源冲突、规则参数非法等自行比对无法发现的问题
• 只有操作返回成功时，代表防火墙规则已经全局同步到所有GCP节点，校验结果比自行查询列表更准确

内容的提问来源于stack exchange，提问作者tester81