函数工作流 FunctionGraph
函数工作流 FunctionGraph
- 最新动态
- 功能总览
- 产品介绍
- 计费说明
- 快速入门
- 用户指南
-
最佳实践
- FunctionGraph最佳实践汇总
- FunctionGraph性能优化实践
- FunctionGraph冷启动优化实践
- FunctionGraph安全最佳实践
- 使用函数处理DIS数据
- 函数+LTS:日志实时分析实战
- 函数+CTS:登录/登出安全分析实战
- 定时开关华为公有云虚拟机
- 使用SpringBoot构建FunctionGraph HTTP函数
- 创建使用自定义认证且后端为FunctionGraph的API
- 函数+APIG:处理文件上传
- 使用函数处理IOT数据
- 函数+DEW:加解密文件
- 工作流+函数:自动化处理OBS中数据
- 函数+LTS:日志实时过滤
- 通过应用中心部署AI绘画Stable-Diffusion
- 使用Go构建FunctionGraph HTTP函数
- 使用FunctionGraph HTTP函数处理gRPC请求
- 使用FunctionGraph部署DeepSeek-R1蒸馏模型
- 开发指南
- API参考
- SDK参考
-
常见问题
-
通用问题
- FunctionGraph是什么
- 使用FunctionGraph是否需要开通计算、存储、网络等服务?
- 使用FunctionGraph开发程序之后是否需要部署?
- 如何获取Token?
- FunctionGraph函数支持哪些编程语言?
- FunctionGraph函数分配磁盘空间有多少?
- FunctionGraph函数是否支持版本控制?
- 函数中如何读写文件?
- 使用CLI工具,如何配置网络代理?
- FunctionGraph函数是否支持扩展?
- IAM子账号使用FunctionGraph需要设置哪些权限?
- 如何制作基于ODBC驱动的Python依赖包用于查询数据库?
- FunctionGraph配额
- FunctionGraph函数支持哪些中文字体?
- FunctionGraph如何实现域名解析?
- 如何通过域名访问专享版APIG中注册的接口?
- 函数工作流的常见使用场景?
- 函数调用绑定在APIG的域名的服务,报域名无法解析?
- 同步函数工作流能否支持到内网最大带宽的同步传输 ?
- 单租户的VPC超过默认配额时,需要怎么做?
- 如何打印info、error、warn级别的日志?
- 函数是否可以把API的接口域名配置成自己的域名?
- 函数工作流是否支持修改运行时语言?
- 已创建的函数是否支持修改函数名称?
- 挂载文件系统时,报“failed to mount exist system path”,应如何处理?
- 如何获取上传的文件?
- 同步调用响应未收到的可能原因?
- os.system("command &")执行日志未采集,应如何处理?
- 自定义运行时,都能操作哪些目录?
- 运行时语言支持的python3.6和3.9具体指哪个版本?
- 用户想使用vpc功能,但不想配置VPC Administrator委托,应配置哪些授权项?
- 函数执行超时的可能原因有哪些?
- 如何获取函数代码?
- 是否有initializer的代码示例?
- 如何开启结构化日志查询
- 函数服务是否支持在函数中启动TCP的监听端口,通过EIP接收外部发送过来的TCP请求?
- FunctionGraph是否支持域名解析?
- 函数发起HTTP请求的源地址如何获取?
- 创建函数
- 触发器管理
- 依赖包管理
-
函数执行
- FunctionGraph函数的执行需要多长时间?
- FunctionGraph函数的执行包含了哪些过程?
- FunctiongGraph函数的并发处理过程是什么?
- FunctiongGraph函数如何处理长时间不执行的实例?
- 首次访问函数慢,如何优化?
- 怎样获取在函数运行过程中实际使用了多少内存?
- 为什么第一次请求会比较慢?
- 调用API时,报错怎么办?
- 如何读取函数的请求头?
- API同步执行函数接口,是否支持内网调用?
- 函数内存超限返回“runtime memory limit exceeded”,如何查看内存占用大小?
- 如何定位自定义镜像执行失败“CrashLoopBackOff”的原因?
- 用户使用相同的镜像名更新镜像,预留实例无法自动更新,会一直使用老镜像,应如何处理?
- 函数配置
- 函数访问外部资源
- 其他问题
- V1迁移V2相关问题
-
通用问题
-
更多文档
-
用户指南(阿布扎比区域)
- 产品介绍
- 快速入门
- 使用前必读
- 构建函数
- 配置函数
- 在线调试
- 配置触发器
- 调用函数
- 监控
- 函数管理
- 依赖包管理
- 预留实例管理
- 扩大资源配额
- 审计
-
常见问题
-
通用问题
- FunctionGraph是什么
- 使用FunctionGraph是否需要开通计算、存储、网络等服务?
- 使用FunctionGraph开发程序之后是否需要部署?
- FunctionGraph函数支持哪些编程语言?
- FunctionGraph函数分配磁盘空间有多少?
- FunctionGraph函数是否支持版本控制?
- 函数中如何读写文件?
- FunctionGraph函数是否支持扩展?
- IAM子帐号使用FunctionGraph需要设置哪些权限?
- 如何制作基于ODBC驱动的Python依赖包用于查询数据库?
- FunctionGraph配额
- FunctionGraph函数如何解析DNS内网域名?
- 容器镜像函数如何解析DNS内网域名?
- 如何通过域名访问专享版APIG中注册的接口?
- 函数工作流的常见使用场景?
- 函数调用绑定在APIG的域名的服务,报域名无法解析?
- 同步函数工作流能否支持到内网最大带宽的同步传输 ?
- 单租户的VPC超过默认配额时,需要怎么做?
- 如何打印info、error、warn级别的日志?
- 函数是否可以把API的接口域名配置成自己的域名?
- 函数工作流是否支持修改运行时语言?
- 已创建的函数是否支持修改函数名称?
- 挂载文件系统时,报“failed to mount exist system path”,应如何处理?
- 如何获取上传的文件?
- 同步调用响应未收到的可能原因?
- os.system("command &")执行日志未采集,应如何处理?
- 自定义运行时,都能操作哪些目录?
- 运行时语言支持的python3.6和3.9具体指哪个版本?
- 用户想使用vpc功能,但不想配置VPC Administrator委托,应配置哪些授权项?
- 函数执行超时的可能原因有哪些?
- 如何获取函数代码?
- 是否有initializer的代码示例?
- 如何开启结构化日志查询
- 函数服务是否支持在函数中启动TCP的监听端口,通过EIP接收外部发送过来的TCP请求?
- 创建函数
- 触发器管理
- 依赖包管理
-
函数执行
- FunctionGraph函数的执行需要多长时间?
- FunctionGraph函数的执行包含了哪些过程?
- FunctiongGraph函数的并发处理过程是什么?
- FunctiongGraph函数如何处理长时间不执行的实例?
- 首次访问函数慢,如何优化?
- 怎样获取在函数运行过程中实际使用了多少内存?
- 为什么第一次请求会比较慢?
- 调用API时,报错怎么办?
- 如何读取函数的请求头?
- 为什么函数实际使用内存大于预估内存,甚至触发OOM?
- 函数内存超限返回“runtime memory limit exceeded”,如何查看内存占用大小?
- 如何定位自定义镜像执行失败“CrashLoopBackOff”的原因?
- 用户使用相同的镜像名更新镜像,预留实例无法自动更新,会一直使用老镜像,应如何处理?
- 函数配置
- 函数访问外部资源
- 其他问题
-
通用问题
- API参考(阿布扎比区域)
- 开发指南(阿布扎比区域)
-
用户指南(吉隆坡区域)
- 产品介绍
- 快速入门
- 使用前必读
- 构建函数
- 配置函数
- 在线调试
- 配置触发器
- 调用函数
- 监控
- 函数管理
- 依赖包管理
- 预留实例管理
- 审计
-
常见问题
-
通用问题
- FunctionGraph是什么
- 使用FunctionGraph是否需要开通计算、存储、网络等服务?
- 使用FunctionGraph开发程序之后是否需要部署?
- FunctionGraph函数支持哪些编程语言?
- FunctionGraph函数分配磁盘空间有多少?
- FunctionGraph函数是否支持版本控制?
- 函数中如何读写文件?
- FunctionGraph函数是否支持扩展?
- IAM子账号使用FunctionGraph需要设置哪些权限?
- 如何制作基于ODBC驱动的Python依赖包用于查询数据库?
- FunctionGraph配额
- 容器镜像函数如何解析DNS内网域名?
- 如何通过域名访问专享版APIG中注册的接口?
- 函数工作流的常见使用场景?
- 函数调用绑定在APIG的域名的服务,报域名无法解析?
- 同步函数工作流能否支持到内网最大带宽的同步传输 ?
- 单租户的VPC超过默认配额时,需要怎么做?
- 如何打印info、error、warn级别的日志?
- 函数是否可以把API的接口域名配置成自己的域名?
- 函数工作流是否支持修改运行时语言?
- 已创建的函数是否支持修改函数名称?
- 挂载文件系统时,报“failed to mount exist system path”,应如何处理?
- 如何获取上传的文件?
- 同步调用响应未收到的可能原因?
- os.system("command &")执行日志未采集,应如何处理?
- 自定义运行时,都能操作哪些目录?
- 运行时语言支持的python3.6和3.9具体指哪个版本?
- 用户想使用vpc功能,但不想配置VPC Administrator委托,应配置哪些授权项?
- 函数执行超时的可能原因有哪些?
- 如何获取函数代码?
- 是否有initializer的代码示例?
- 如何开启结构化日志查询
- 函数服务是否支持在函数中启动TCP的监听端口,通过EIP接收外部发送过来的TCP请求?
- 创建函数
- 触发器管理
- 依赖包管理
-
函数执行
- FunctionGraph函数的执行需要多长时间?
- FunctionGraph函数的执行包含了哪些过程?
- FunctiongGraph函数的并发处理过程是什么?
- FunctiongGraph函数如何处理长时间不执行的实例?
- 首次访问函数慢,如何优化?
- 怎样获取在函数运行过程中实际使用了多少内存?
- 为什么第一次请求会比较慢?
- 调用API时,报错怎么办?
- 如何读取函数的请求头?
- 为什么函数实际使用内存大于预估内存,甚至触发OOM?
- 函数内存超限返回“runtime memory limit exceeded”,如何查看内存占用大小?
- 如何定位自定义镜像执行失败“CrashLoopBackOff”的原因?
- 用户使用相同的镜像名更新镜像,预留实例无法自动更新,会一直使用老镜像,应如何处理?
- 函数配置
- 函数访问外部资源
- 其他问题
-
通用问题
- 修订记录
- API参考(吉隆坡区域)
- 开发指南(吉隆坡区域)
-
用户指南(阿布扎比区域)
- 通用参考
本文导读
展开导读
链接复制成功!
deploy.py代码示例
代码示例
以下为自动化部署deploy.py文件的代码示例。
该示例用于自动化部署和更新华为云FunctionGraph函数,涵盖配置和代码的更新。通过解析配置文件,调用命令行工具执行更新命令,并处理加密数据的解密,同时记录日志以确保操作的可追溯性。可参考代码注释使用。
# -*-coding:utf-8 -*-
import os
import sys
import json
import logging
import subprocess
from yaml import load
from base64 import b64decode
from Crypto.Cipher import AES
# need: pip install pyyaml
try:
from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
from yaml import Loader, Dumper
logging.basicConfig(level=logging.INFO,
filename='function.log',
filemode='a',
format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s')
def decrypt(json_input, key):
# We assume that the key was securely shared beforehand
try:
b64 = json.loads(json_input)
json_k = ['nonce', 'header', 'ciphertext', 'tag']
jv = {k: b64decode(b64[k]) for k in json_k}
cipher = AES.new(key.encode(), AES.MODE_GCM, nonce=jv['nonce'])
cipher.update(jv['header'])
plaintext = cipher.decrypt_and_verify(jv['ciphertext'], jv['tag'])
return plaintext.decode()
except (ValueError, KeyError) as e:
raise e
def generate_update_function_config_cmd(new_config, old_config, key):
# 函数执行入口
handler = new_config['handler']
# 函数runtime配置(必填但不支持修改)
runtime = new_config['runtime']
# 函数内存规格配置
memory_size = new_config['memorySize']
# 函数执行超时配置
timeout = new_config['timeout']
# 函数所属project_id
project_id = new_config['projectID']
# 拼装更新函数配置命令
update_cmd = f'hcloud FunctionGraph UpdateFunctionConfig' \
f' --cli-region="{region}"' \
f' --function_urn="{function_urn}"' \
f' --project_id="{project_id}"' \
f' --handler="{handler}"' \
f' --timeout={timeout}' \
f' --memory_size={memory_size}' \
f' --runtime="{runtime}"' \
f' --func_name="{function_name}"'
# 用户环境变量配置
# 更新用户环境变量为直接覆盖,如果有手动在函数界面配置环境变量没有更新到cam.yaml文件内
# 则手动添加环境变量配置则丢失
user_data = new_config.get('userData', None)
if user_data is not None:
user_date_json_str = json.dumps(user_data)
user_date_json_str = json.dumps(user_date_json_str)
update_cmd = update_cmd + f' --user_data={user_date_json_str}'
encrypted_user_data = new_config.get('encryptedUserData', None)
if encrypted_user_data is not None:
encrypted_user_data = decrypt(encrypted_user_data, key)
encrypted_user_date_json_str = json.dumps(encrypted_user_data)
update_cmd = update_cmd + \
f' --encrypted_user_data={encrypted_user_date_json_str}'
# 如果有vpc则保留
vpc_config = old_config.get('func_vpc', None)
if vpc_config is not None:
update_cmd = update_cmd + \
f' --func_vpc.vpc_name={vpc_config["vpc_name"]}' \
f' --func_vpc.vpc_id={vpc_config["vpc_id"]}' \
f' --func_vpc.subnet_id={vpc_config["subnet_id"]}' \
f' --func_vpc.cidr={vpc_config["cidr"]}' \
f' --func_vpc.subnet_name={vpc_config["subnet_name"]}' \
f' --func_vpc.gateway={vpc_config["gateway"]}'
# 如果有委托配置则保留 "xrole": "function-admin"和"app_xrole": "function-admin",
xrole_config = old_config.get('xrole', None)
if xrole_config is not None:
update_cmd = update_cmd + f' --xrole="{xrole_config}"'
app_xrole_config = old_config.get('app_xrole', None)
if app_xrole_config is not None:
update_cmd = update_cmd + f' --app_xrole="{app_xrole_config}"'
# 配置初始化入口和初始化超时时间
initializer_handler = new_config.get('initializerHandler', None)
initializer_timeout = new_config.get('initializerTimeout', None)
if initializer_handler is not None and initializer_timeout is not None:
update_cmd = update_cmd + \
f' --initializer_handler="{initializer_handler}" ' \
f'--initializer_timeout={initializer_timeout}'
# 并发配置
strategy_config = new_config.get('strategyConfig', None)
if strategy_config is not None:
concurrency = strategy_config.get('concurrency', None)
# 单实例并发数
concurrent_num = strategy_config.get('concurrentNum', None)
update_cmd = update_cmd + \
f' --strategy_config.concurrency="{concurrency}" ' \
f'--strategy_config.concurrent_num={concurrent_num}'
# 如果有磁盘挂载则保留
mount_config = old_config.get('mount_config', None)
if mount_config is not None:
mount_user = mount_config["mount_user"]
update_cmd = update_cmd + \
f' --mount_config.mount_user.user_id={mount_user["user_id"]}' \
f' --mount_config.mount_user.user_group_id={mount_user["user_group_id"]}'
func_mounts = mount_config["func_mounts"]
i = 1
for func_mount in func_mounts:
update_cmd = update_cmd + \
f' --mount_config.func_mounts.{i}.mount_resource="{func_mount["mount_resource"]}"' \
f' --mount_config.func_mounts.{i}.mount_share_path="{func_mount["mount_share_path"]}"' \
f' --mount_config.func_mounts.{i}.mount_type="{func_mount["mount_type"]}"' \
f' --mount_config.func_mounts.{i}.local_mount_path="{func_mount["local_mount_path"]}"'
i = i + 1
return update_cmd
def exec_cmd(cmd):
proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
outs, _ = proc.communicate()
return outs.decode('UTF-8')
def check_result(stage, exec_result):
if "USE_ERROR" in exec_result:
error_info = f"failed to {stage}: {exec_result}"
logging.error(error_info)
raise Exception(error_info)
if "FSS.0409" in exec_result:
error_info = f"failed to {stage}: {exec_result}"
logging.error(error_info)
# 返回错误为函数代码没变不需要更新错误则返回
return
try:
result_object = json.loads(exec_result)
except Exception:
error_info = f"failed to {stage}: {exec_result}"
logging.error(error_info)
raise Exception(error_info)
if "error_code" in result_object:
error_message = result_object["error_msg"]
error_info = f"failed to {stage}: {error_message}"
logging.error(error_info)
raise Exception(error_info)
def generate_update_function_code_cmd():
cmd = \
f'hcloud FunctionGraph UpdateFunctionCode --cli-region="{region}"' \
f' --function_urn="{function_urn}" --project_id="{project_id}"' \
f' --code_url="{code_url}" --func_code.link="" --func_code.file="" --code_type="obs" '
depend_list = old_function_code.get("depend_list", None)
if depend_list is not None and len(depend_list) > 0:
i = 1
for depend_id in depend_list:
cmd = cmd + f'--depend_list.{i}="{depend_id}"'
return cmd
if __name__ == '__main__':
deploy_function_path = sys.argv[1]
key = sys.argv[2]
f = open(os.path.join(deploy_function_path, "cam.yaml"))
data = load(f, Loader=Loader)
function_config = data['components'][0]
function_name = function_config['name']
function_properties = function_config['properties']
region = function_properties['region']
code_url = function_properties['codeUri']
project_id = function_properties['projectID']
# 拼接获取函数urn
function_urn = "urn:fss:" + region + ":" + project_id + \
":function:default:" + function_name + ":latest"
logging.info(f"start to deploy functionURN:{function_urn}")
# 查询函数的配置信息
query_function_config_cmd = \
f'hcloud FunctionGraph ShowFunctionConfig --cli-region="{region}"' \
f' --function_urn="{function_urn}" --project_id="{project_id}"'
result = exec_cmd(query_function_config_cmd)
# 主要是查看函数是否有配置VPC和委托,如果有更新函数配置时需要带上,避免更新导VPC或委托配置丢失
old_function_config = json.loads(result)
check_result("query function config", result)
# 查询函数代码,主要是函数绑定依赖包信息保留
query_function_code_cmd = \
f'hcloud FunctionGraph ShowFunctionCode --cli-region="{region}"' \
f' --function_urn="{function_urn}" --project_id="{project_id}"'
result = exec_cmd(query_function_code_cmd)
old_function_code = json.loads(result)
logging.info("query function %s code result: %s", function_urn, result)
check_result("query function code", result)
# 更新函数代码
query_function_code_cmd = generate_update_function_code_cmd()
result = exec_cmd(query_function_code_cmd)
logging.info("update function %s code result: %s", function_urn, result)
check_result("update function code", result)
# 更新函数配置
update_function_config_cmd = generate_update_function_config_cmd(
function_properties, old_function_config, key)
result = exec_cmd(update_function_config_cmd)
logging.info("update function %s config result: %s", function_urn, result)
check_result("update function config", result)
logging.info(f"succeed to deploy function {function_urn}")
父主题: 自动化部署