Prefect Deployment CRD

[marcm-ml]

I think this operator could also manage Prefect Deployments.
The operator would manage deployments by creating, updating and deleting them.
This would also aid managing prefect deployment in GitOps workflows like flux/argo. The resource could implement a similar interface to flux's HelmRelease:

apiVersion: prefect.io/v1
kind: PrefectDeployment
metadata:
  labels:
    app.kubernetes.io/name: prefect-operator
    app.kubernetes.io/managed-by: kustomize
  name: myDeployment
status:
  deploymentID: "f31ba293-4028-4492-9219-c4eddf56707e"
  flowID: "e1e69e4b-f767-445e-8cd0-900a3dcbfc2f"
spec:
  interval: 10m # reconcile interval
  deletionPolicy: "delete"
  retryInterval: 5m # retry if reconcile fails
  prefectApiUrl: "http://prefect-server:4200/api"
  deploymentSpec:
    schedule: "0 * * * *"
    entrypoint: "path/to/flow:func"
    storage: None
    tags: 
      - myTag
    jobVariables:
       image: "myimage:latest"
       labels:
         myLabel: myValue
       command: ["prefect.engine"]
       ...

What do you think?

[chrisguidry]

Hi @marcm-ml, that sounds like a great idea! Are you interested in building that?

Should nobody picks this up in a few weeks, I wouldn't mind helping with the delivery of this feature if the technical specifications are clearly laid out. To start off with the specification of the CRD for a PrefectDeployment. I assume the specification for the CRD should largely reflect the behavior of the api contract for POST /api/deployments. Is this a reasonable assumption? If not, could you suggest any appropriate documentation or stipulations to this?

Also, how if at all would you expect the prefect-kubernetes functionality in the the SDK to be incorporated into the creation of a CRD like this? Do you anticipate the CRD trying to mostly behave as a wrapper for the SDK?

[marcm-ml]

Thanks for the interest. I most likely wont find the time to implement this feature on my own. But I can assist @josephmcasey with the technical specs.

I think for understanding where I am coming from, it makes sense to lay out my current workflow for deploying prefect deployments via fluxCD:

Details

I have many different repository varying in size and scope. Each repo may contain one or more flows, lets assume every repo has one or more flow.py files. Each flow.py contains the flow and the deployment definition. For ease of use, it also contains a CLI. The RunnerDeployment class uses the prefect client client.create_flow_from_name and client.create_deployment API endpoint (see here), which is the same you suggested. All in all, this looks something like this:

from typing import Annotated, Optional

import typer
from prefect import flow
from prefect.deployments.runner import RunnerDeployment

app = typer.Typer()


@flow()
def my_flow(dry_run: bool = False):
    ...

@app.command()
def deploy(dry_run: Annotated[bool, typer.Option()] = False):
    # this could be any runner compatible Storage interface
    from prefect.runner.storage import LocalStorage
    storage = LocalStorage(path="/absolute/path/to/dir")

    deployment = RunnerDeployment.from_flow(
        name='my_deployment',
        job_variables={},
        ...
    )
    deployment.storage = storage

    # useful for debugging
    print("Created deployment with this configuration:")
    print(dict(prefect_deployment))

    # do not commit to prefect API when dry-running
    if not dry_run:
        print("Sending deployment to PREFECT API")
        prefect_deployment.apply()
    else:
        print("Running in dry-mode. Not uploading...")

    print(f"{prefect_deployment.flow_name}/{prefect_deployment.name} deployed")
    print("")


if __name__ == '__main__':
    app()

Any source code, including flow.py, is copied into the docker image which I need to build and publish in order to automate the deployment of the API within my kubernetes cluster.

---

Getting this into the cluster and API, works as follows:

Within each repo there is a flux directory laid out as follows:

flux
├── kustomization.yaml
└── prefect-deployer.yaml

The prefect-deployer is a kubernetes Job that uses the docker image and executes the "deploy" command form the script above. This creates/updates the deployment within prefect.

I spare you the details but Flux takes care of creating this kubernetes job periodically looking up the definitions within this directory using some flux CRDs (GitRepository and Kustomization).
The prefect job is always reconciled if something changes within its definition OR the job failed. This guarantees me that eventually I have consistency with what is defined in code in my repo and what is defined within the Prefect API.

You could argue that you can use a CICD (push GitOps) for all of this. However, my workflow entirely relies on fluxCD (pull GitOps) and I don't want to break this pattern with my prefect flows. Also using CICD, you lose the ability automatically react on changes within the prefect API, i.e. if somebody manually changed the deployment and it needs reconciliation.

Side note: I have not yet played with the flow.serve() or flow.deploy() commands. The first time I looked at this, it was not something that could be used for my purposes.

---

Yes, the prefect Deployment CRD API should reflect the Prefect Deployment API endpoint but also the Prefect Flow API endpoint for creating the Flow as the Deployment API would not be sufficient imo. I think it is enough to take a look at the RunnerDeployment class and model the same behaviour.

I think as an initial start the CRD would be created manually or at least translated from a prefect.yaml to the CRD definition. However, in the end I would expect some function to generate the CRD from a flow definition either via Python code (similar to the RunnerDeployment) or via Prefect CLI (similar to how this can be for the prefect.yaml file). This extended functionality could be implemented in prefect-kubernetes as an add-on to the core prefect functionality.

Operator Pseudo Code

Within the operator's watch loop do these steps:

• Create/Update (can be combined since prefect server takes care of creating/updating)
  1. Read CRD
  2. Translate flow specs into Prefect's Flow API compliant JSON
  3. Call create_flow endpoint
  4. Translate deployment specs into Prefect's Deployment API compliant JSON
     • this could include translating a custom storage spec into proper PullSteps, just like it is done here for the PrefectDeployment
  5. Call create_deployment API endpoint with Flow ID from response in step 3.
  6. Update status with flow and deployment IDs or error status
• Delete
  1. Check Deployment deletionPolicy, and exit if deletionPolicy is "retain"
  2. Read Deployment ID from status
  3. Call delete_deployment API endpoint with Deployment ID
• Reconciliations
  1. Check periodically (reconcileInterval param) if the deployment needs reconciliation
  2. Call get_deployment endpoint
  3. Read deployment spec
  4. Compare with CRD spec
  5. call create/update
• Error Reconciliation
  1. call create/update again within retryInterval param

If you need more info, feel free to ask anytime.

[chrisguidry]

That makes total sense, @marcm-ml! I also have a need to deploy Prefect flows via FluxCD, this would be optimal for me too. Nice work on your Python + K8s Job solution, that's not bad, but a CRD with operator support would be ideal.

@josephmcasey amazing, let's collaborate on this! Like Marc is pointing out, creating a Prefect deployment requires upserting both a Flow and a Deployment. There are tons of examples of this in the Prefect client SDK (in Python, so not directly usable here). Prefect's Deployment schema from the API is the foundation, but it has also gone through quite a few iterations over the last few years. There are definitely parts of it that only remain for backwards compatibility with versions of Prefect 2, which we can streamline to focus on Prefect 3. Happy to help guide that pruning, it shouldn't be too complex.

But fundamentally, yes, we'd want to follow the API for the most part. The spec will need to leave space for other things, too, like the reference to the PrefectServer or external API (as we do with work pools) and the work pool to deploy to.

apiVersion: prefect.io/v1
kind: PrefectDeployment
metadata:
  # the name here would become the name on Prefect as well
  name: the-deployment
spec:
  # The `server` would be a `PrefectServerReference` like with work pools
  server:
    remoteApiUrl: https://api.prefect.cloud
    accountId: aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa
    workspaceId: bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb
    apiKey:
      valueFrom:
        secretKeyRef:
          name: the-api-key-secret
          key: the-api-key

  workPool:
    namespace: some-namespace
    name: my-work-pool
    workQueue: some-work-queue
  
  # here is where the deployment details should plug in
  deployment:
    ...

status:
  id: "dddddddd-dddd-dddd-dddd-dddddddddddd"
  flow_id: "ffffffff-ffff-ffff-ffff-ffffffffffff"

The spec of the deployment itself is a great topic for discussion. We can start for sure with all of the API fields as given, minus some legacy things like the *_document_id fields. First up, we need a reference to a work pool, which I think makes the most sense as a namespaced object reference:

spec:
  deployment:


    # about the deployment
    description: ...
    tags:
      - tag-1
      - tag-2
    labels:
      key: value1
      key2: 42
    version_info:
      type: "prefect:simple"
      version: abc123


    # about the code to run
    entrypoint: "my_code.py:my_function"
    path: /where/is/this/code
    pull_steps:
      - ...change directory...
      - ...pull from git...


    # parameters to the flow run
    parameter_openapi_schema: {}  # this would need to be generated
    enforce_parameter_schema: true
    parameters: {}
    job_variables: {}


    # whether and when to run it
    paused: false
    schedules: 
      - slug: every-two-minutes
        schedule:
          interval: 120
          anchor_date: "2020-01-01T00:00:00Z"
          timezone: "America/New_York"
        active: true
        max_scheduled_runs: 1
        parameters: {}


    # concurrency limiting
    concurrency_limit: 123
    global_concurrency_limit:
      active: true
      name: my-global-concurrency-limit
      limit: 123
      slot_decay_per_second: 2
    concurrency_options:
      collision_strategy: ENQUEUE

As I'm compiling these, I realize that we may be asking a lot of the user to plug in all of the things like the parameter_openapi_schema and pull_steps by hand. Maybe we should farm part of this out to the Prefect client SDK?

[mitchnielsen]

Just a quick side note for posterity or anyone else looking: as an alternative here, you can manage Prefect deployments with Terraform (docs), meaning the Prefect Operator can be used together with the Prefect Terraform provider.

Obviously if Terraform isn't in the desired tech stack, managing them with the Operator sounds like a nice solution 👌🏼

[marcm-ml]

I think a work pool and PrefectApiUrl should be either provided as a namespaced reference or simply as string. So there is the option to not use operator managed instances.

Good point, the parameter schema is problematic from UX perspective. I believe this might be something that the prefect CLI/SDK can provide programmatically which the user can then access. Of course a command that would generate a complete CRD spec is more desirable.

I believe the pull steps are also difficult to provide manually.

Maybe as an intermediate step, you could provide a script that takes the json of the runner deployment and translate it to the CRD spec. Again I think the end-goal should be that you directly generate the CRD from the python Prefect CLI/SDK

Initially, we could get away with a simple script that outputs the correct things from the current prefect SDK. My current solution/script using the RunnerDeoyment class already generates all the necessary parameters. Not ideal UX-wise but I would not see this as a blocker for the CRD

[chrisguidry]

I like that approach @marcm-ml, we can make a prefect CLI utility that does this eventually, but a small one-off script in prefect-operator can work for the near term. And folks who have deployments saved in their prefect.yaml already can just use the values there as they see fit.

[chrisguidry]

Hey folks, I've got a PR up that covers the lion's share of this: #170. I also included the Python script to produce the K8s manifest like we discussed

[chrisguidry]

One note: I ended up sticking with camelCase instead of snake_case for the deployment fields for consistency with how most things work in K8s.

[chrisguidry]

https://github.com/PrefectHQ/prefect-operator/pull/170/files#diff-5b70b95904fa7a6a813981cab878e5dbe6e662a94e1ea1bbbbfaa1ded04f0776 has an example deployment:

apiVersion: prefect.io/v1
kind: PrefectDeployment
metadata:
  name: hello-there
spec:
  server:
    name: prefect-ephemeral

  workPool:
    name: process-pool

  deployment:
    entrypoint: "flows/hello_world.py:hello"
    pullSteps:
      - prefect.deployments.steps.git_clone:
          repository: https://github.com/PrefectHQ/examples.git
          branch: main
    parameters:
      name: "prefect-operator!"
    parameterOpenApiSchema:
      title: Parameters
      type: object
      properties:
        name:
          default: Marvin
          position: 0
          title: name
          type: string
      required: []
      definitions: {}