Запуски Prefect Flow: как отправить журналы запуска на сервер префекта?

В настоящее время я пытаюсь запустить несколько префектов Flow на своем локальном компьютере (затем в некоторых экземплярах Google Cloud Run), и я хотел бы отправлять журналы запусков на удаленный префект-сервер.

Вот пример кода:

@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 ' Мой первый поток», уровень: INFO....

Мне интересно, что я забыл или что я делаю неправильно :(

Спасибо за помощь


cloud flow orchestration prefect
person AlaeddineB    schedule 22.12.2020    source источник

Ответы (1)


arrow_upward
0
arrow_downward

В экосистеме Prefect есть два типа прогонов:

  • интерактивные прогоны: это тот тип прогона, который вы здесь пытаетесь выполнить; это запуск, который вы создаете в своем локальном процессе и отвечаете за управление/мониторинг. Интерактивные прогоны не знают, как общаться с Prefect API, и не пытаются это делать. Эти запуски в основном предназначены для локального тестирования.

  • управляемые прогоны: это прогоны, о которых API Prefect знает, отслеживает и координирует. Вы можете создавать эти прогоны только через API Prefect, либо через расписание в зарегистрированном потоке, либо через специально созданный поток. run (это вызов API).

Ваш код пытается смешать эти два; вы видите эту ошибку, потому что созданный вами запуск не имеет соответствующей информации для связи с серверной частью. Я предлагаю ознакомиться с некоторыми учебниками Prefect, возможно, начав с этого.

person chriswhite    schedule 07.01.2021
comment
Спасибо за ответ, но по вашей ссылке приведены инструкции по работе с Prefect Cloud. Я думаю, что решение моей проблемы состоит в том, чтобы запустить локальный агент и запустить поток на этом агенте. Я работаю над этим.. - person AlaeddineB; 11.01.2021
comment
В этом руководстве рассматриваются шаги для облака или сервера, включая запуск агента. Любые шаги, отмеченные идентификатором облака, можно пропустить, если вы используете Сервер. - person chriswhite; 11.01.2021