Architectural guidance for strict 1:1 mapping of agent and workflow instances; scheduling

[alexanderwiller]

Hello everybody,

currently I am exploring Prefect for automation of various IT processes. To start with, I am implementing a solution for backup of a minecraft server, being very manageable, as it consists only of a few distinct steps.

The current architectural approach is as follows:

• Write Task and Workflow definitions and package them as a Docker image with a bit of plumbing code
• Use the image for a k8s sidecar container attached to the minecraft server container
• Configure the codified Workflow/Tasks "template" by means of environment variables set via k8s, e.g. the Cron schedule and other parameters
• Let the workflow register itself at Prefect server on sidecar start and then go into agent mode to wait for execution requests
• Execute all Workflow Tasks on the same host where the dedicated local agent is running
• Use idempotency_key=flow.serialized_hash() to bump Workflow version if any parameter has been changed

The idea behind this is to allow different minecraft server deployments to have the same workflow structure and tasks, but with different parameters - basically, to create reusable workflows. Registration is meant to provide insight into the process and to allow monitoring and manual workflow execution.

I recognize that this may not be the most common arrangement of agents and workflows, but to me a still a very useful one.

My primary question is what would be the best way to map Agents and per-Agent Workflows 1:1 to each other? I have seen that a host label is added to Agents (and probably workflows?) automatically, which would be the most basic mapping approach. It shall be strictly avoided to mix non-paired Agents and Workflows, so I also considered generating a UUID for this 1:1 match. Is there any definite approach to do this?

In addition, I am somewhat unsure which role the Prefect server plays in Cron scheduling. Its clear that the schedule itself is defined as part of the workflow (instance) - which is desirable, as it tightly coupled to the concrete k8s-managed application. I have already observed that the core library is able to work completely independent of any server, just executing the flow as scheduled.

How does this differ when the workflow is not run(), but registered to the server, followed by start of the local Agent waiting for work to do?

I'd highly appreciate any advice 😃

For reference, I've also attached the structurally very simple code:

import prefect
from prefect import task, Flow, resource_manager
from prefect.tasks.shell import ShellTask
from prefect.schedules import Schedule
from prefect.schedules.clocks import CronClock
from prefect.agent.local import LocalAgent
from prefect.run_configs import LocalRun

import argparse
import uuid
import os
import socket
from mcipc.rcon.je import Client

# PREFECT__CLOUD__AGENT__AUTH_TOKEN
# PREFECT__CLOUD__API
# PREFECT__BACKEND
# PREFECT__CLOUD__AGENT__AGENT_ADDRESS ???

PREFECT_FLOW_NAME       = os.environ.get('FLOW_NAME', 'minecraft-backup')
PREFECT_PROJECT_NAME    = os.environ.get('PREFECT_PROJECT_NAME', 'default')
PREFECT_BACKEND_URL     = os.environ.get('PREFECT_BACKEND_URL', None)
CRON_SCHEDULE           = os.environ.get('CRON_SCHEDULE', '0 6 * * *')
RCON_HOST               = os.environ.get('RCON_HOST', '127.0.0.1')
RCON_PORT               = os.environ.get('RCON_PORT', 25575)
RCON_PASSWORD           = os.environ.get('RCON_PASSWORD', 'minecraft')
RCON_TIMEOUT            = os.environ.get('RCON_TIMEOUT', 15)
DATA_DIR                = os.environ.get('DATA_DIR', '/data')
RESTIC_TAGS             = os.environ.get('RESTIC_TAGS', []) + [PREFECT_FLOW_NAME, socket.gethostname()]

parser = argparse.ArgumentParser()
parser.add_argument('action', choices=['agent', 'run'])
args = parser.parse_args()

### Task and flow definition

@resource_manager(
    setup_task_kwargs={'timeout': RCON_TIMEOUT},
    cleanup_task_kwargs={'timeout': RCON_TIMEOUT}
)
class MinecraftSaveDisabledContext:
    def setup(self):
        logger = prefect.context.get("logger")
        with Client(RCON_HOST, RCON_PORT, passwd=RCON_PASSWORD) as client:
            logger.info(client.save_off())

    def cleanup(self, resource):
        logger = prefect.context.get("logger")
        with Client(RCON_HOST, RCON_PORT, passwd=RCON_PASSWORD) as client:
            logger.info(client.save_on())

def setup_flow() -> Flow:
    shell_task = ShellTask(return_all=True)

    with Flow("minecraft-backup") as flow:
        with MinecraftSaveDisabledContext():
            shell_task(command=f"restic --verbose backup {DATA_DIR}")

    flow.schedule = Schedule(clocks=[CronClock(CRON_SCHEDULE)])
    flow.run_config = LocalRun()
    return flow

### Registration, agent operation, immediate execution

flow = setup_flow()
# print(prefect.utilities.diagnostics.flow_information(flow))

if args.action == 'agent':
    flow.register(
        project_name=PREFECT_PROJECT_NAME,
        idempotency_key=flow.serialized_hash()
    )
    LocalAgent(labels=["TODO"]).start()
elif args.action == 'run':
    flow.run()
else:
    assert False

[cicdw]

Hi @alex04103 ! First off, I think your use case sounds really interesting - if you ever would consider writing a blog post on it, I think that could be really fun for folks!

To your questions:

For specifying flow / agent relationships, labels are definitely the way to go; labels can be any string, so whether it's UUIDs or a simple name is up to you - agents that have a superset of your flow's labels will be able to run them.

Now, it appears that the only thing that changes from flow to flow is the DATA_DIR. If it were me, I would make that a Prefect Parameter and actually schedule its value to change on whatever clock makes sense. In addition, you can schedule labels for your individual runs so that they are sent to different agents with different parameter values -- and you can do all of this with a single registered flow!

For your larger question about flow.run vs. a registered flow: flow.run is a fully interactive workflow run; as you observed, your schedule will be respected, etc. For certain use cases, this is perfectly sufficient. However, many users end up requiring fault tolerant execution and scheduling, multiple flows to manage, a monitoring dashboard, a persistent state layer for tracking history and logs, notifications, etc. This is where the Prefect APIs come into play - Prefect Server is generally useful for small scale, single node monitoring of workflows where users are comfortable managing some infrastructure themselves; Prefect Cloud is best suited for high availability, distributed runs, and comes with more guarantees, automation features and scalability (and of course for those users who would rather not manage and host an API and metadata layer themselves, Cloud requires essentially 0 setup).