Add BaseWorker and ProcessWorker

[desertaxle]

This PR adds a BaseWorker class to implement most of the mechanics and patterns needed for workers. BaseWorker is an abstract base class that implements storage scanning, queue polling, and sending worker heartbeats.

Workers scan a workflow storage location at a given interval and apply any discovered deployments so that scheduled flow runs can be executed by the worker pool that the deployment was submitted to. Currently, the way to submit a deployment to a worker is by placing the deployment manifest and flow code into the worker's configured workflow storage location. We plan on creating convenience methods for submitting deployments to workers so that the mechanics of where the deployment manifest and flow code need to be submitted is abstracted.

Implementations of BaseWorker are required to implement a .run method which is used for executing a flow run. This run method is roughly analogous to the .run method on an infrastructure block. BaseWorker implementation are also required to implement a .verify_submitted_deployment method. The .verify_sbumitted_deployment method gives each worker implementation an opportunity to verify that the deployment was submitted to the correct worker type and the current worker will be able to execute scheduled flow runs for the submitted deployment.

Workers are capable of creating a worker pool if the worker pool that they are configured with at start up does not already exist. Otherwise, workers will warn if they join a worker pool that is configured for a different worker type. This functionality can be further expanded to ensure worker compatibility with worker pools that they are joining.

This PR also adds a ProcessWorker which is the first BaseWorker implementation. It is largely a port of the Process infrastructure block.

Features for workers that are necessary, but will be implemented in another PR:

• Flow run cancellation
• Run customization via base job configurations and job variables (roughly analogous to attributes on infrastructure blocks)
• Specification of worker pool name and worker pool queue name when creating a deployment

All of this functionality is gated behind the PREFECT_EXPERIMENTAL_ENABLE_WORKERS setting.

Example

Start a worker:

# Enable the experimental workers feature
prefect config set PREFECT_EXPERIMENTAL_ENABLE_WORKERS=1

prefect worker start --name test-worker --pool test-worker-pool --type process

To submit a deployment:

• Start a worker
• Build a deployment with prefect deployment build ...
• Add worker_pool_name to the generated deployment manifest
  • e.g.

     ...
    description: null
    version: 5785d5ce817175647f71a35d8bd8df73
    worker_pool_name: test-worker-pool
    tags: []
    ...

• Move the deployment manifest and flow code to PREFECT_WORKER_WORKFLOW_STORAGE_PATH (defaults to ~/.prefect/workflows)
• Deployment should show up in the UI within ~30 seconds

Checklist

• This pull request references any related issue by including "closes <link to issue>"
  • If no issue exists and your change is not a small fix, please create an issue first.
• This pull request includes tests or only affects documentation.
• This pull request includes a label categorizing the change e.g. fix, feature, enhancement

[netlify]

✅ Deploy Preview for prefect-orion ready!

Name	Link
🔨 Latest commit	b4d410b
🔍 Latest deploy log	https://app.netlify.com/sites/prefect-orion/deploys/63b8e1e955bef7000821c86e
😎 Deploy Preview	https://deploy-preview-7996--prefect-orion.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify site settings.

[zanieb]

What if it's an exception?

[desertaxle]

The exception is handled in _submit_run_and_capture_errors where the flow run is either marked as failed if submission failed, or a error is logged saying that something might have gone wrong. This check is to make sure that the worker doesn't try to update the flow run in the backend if something when wrong during submission.

This was taken from the existing agent implementation so this could be a good time to improve this logic if there's a better way to do it.

[cicdw]

I'm sure we'll learn more as we work with this, but this LGTM

[zanieb]

@desertaxle Please don't merge until 2.7.7 is out.