问题描述
我目前正在尝试在我的本地机器上运行一些省长流程(然后在一些 Google 云运行实例中),我想将运行日志发送到远处的省长服务器。
这是示例代码:
@task
def my_task():
logger = prefect.context.get("logger")
logger.info("An info message.")
logger.warning("A warning message.")
@task(log_stdout=True)
def say_hello(name):
print("Hello,{}!".format(name))
with Flow(name="My First Flow") as flow:
my_task()
say_hello(Parameter('msg'))
flow.run(msg='You')
另外,我在 backend.toml 中加入了一些配置:
backend = "server"
[server]
host = "SERVER_PUBLIC_IP"
port = "4200"
host_port = "4200"
endpoint = "${server.host}:${server.port}"
[server.graphql]
host = "SERVER_PUBLIC_IP"
[server.ui]
host = "SERVER_PUBLIC_IP"
port = "8080"
host_port = "8080"
endpoint = "${server.ui.host}:${server.ui.port}"
graphql_url = "SERVER_PUBLIC_IP:4200/graphql"
[logging]
# The logging level: NOTSET,DEBUG,INFO,WARNING,ERROR,or CRITICAL
level = "INFO"
# The log format
format = "[%(asctime)s] %(levelname)s - %(name)s | %(message)s"
# additional log attributes to extract from context
# e.g.,log_attributes = "['context_var']"
log_attributes = "[]"
# the timestamp format
datefmt = "%Y-%m-%d %H:%M:%s"
log_to_cloud = true
所以,这是我遇到的问题:
INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at
"input.logs[0].flow_run_id"; Expected non-nullable type UUID! not to be null.
The GraphQL query was:
mutation($input: write_run_logs_input!) {
write_run_logs(input: $input) {
success
}
}
因为我的日志的第一行没有 flow_run_id。
示例: {"input": {"logs": [{"flow_run_id": null,"task_run_id": null,"timestamp": "2020-12-22T21:24:13.628595+00:00","name" : "prefect.FlowRunner","message": "Beginning Flow run for 'My First Flow'","level": "INFO"....
我想知道我忘记了什么或者我做错了什么:(
感谢帮助
解决方法
Prefect 生态系统中有两种类型的运行:
-
交互式跑步:这是您在此处尝试的跑步类型;这是您在本地进程中生成并负责管理/监控的运行。交互式运行不知道如何与 Prefect API 对话,它们也不会尝试。这些运行主要用于本地测试。
-
托管运行:这些是 Prefect API 了解、跟踪和编排的运行 - 您只能通过 Prefect API 创建这些运行,或者通过注册流程上的计划或通过临时创建的运行(这是一个 API 调用)。
您的代码试图混合这两者;您看到该错误是因为您创建的运行没有与后端通信的适当信息。我建议查看一些 Prefect 教程,也许从 this one
开始