PrefectDeployment: allow overriding the derived flow name (and deployment name) to match existing deployments

[sghegoiu-rokos]

My apologies in advance for the AI generated description, but it did a good job at articulating the issue.

Summary

PrefectDeployment derives the Prefect flow name verbatim from the entrypoint token and the
Prefect deployment name from the Kubernetes object name, with no way to override either. This
makes it impossible to reconcile a CR onto an existing deployment whose flow/deployment names don't
happen to match those derived values — the operator instead creates a new, duplicately-named flow, and
breaks any consumer that resolves deployments by name.

Two small, optional, backward-compatible fields on spec.deployment would fix this:
flowName and name.

Current behavior

1. Flow name is the substring after : in the entrypoint, used verbatim to create/get the flow:

   internal/prefect/convert.go (GetFlowIDFromDeployment):

   _, after, ok := strings.Cut(entryPoint, ":")
   ...
   flowName := after
   flowSpec := &FlowSpec{ Name: flowName, ... }
   flow, err := client.CreateOrGetFlow(ctx, flowSpec)

2. Deployment name is forced to the k8s object name:

   internal/prefect/convert.go (ConvertToDeploymentSpec):

   spec := &DeploymentSpec{ Name: k8sDeployment.Name, FlowID: flowID }

So for an entrypoint flows/foo.py:run_foo on a CR named my-deployment, the operator always
produces deployment my-deployment under flow run_foo.

Why this is a problem

When a flow is authored with a decorator name, e.g.:

@flow(name="My Nicely Named Flow")
def run_foo(): ...

prefect deploy registers the deployment under the flow My Nicely Named Flow (it imports the
flow and reads flow.name). The operator, which intentionally does not import Python, registers it
under run_foo instead. Result:

• Duplicate flows. Migrating an existing prefect deploy-created deployment to the operator
  creates a second flow (run_foo) rather than reconciling onto the original (My Nicely Named Flow).
• By-name resolution breaks. Anything that resolves a deployment via
  GET /api/deployments/name/{flow_name}/{deployment_name} or run_deployment("Flow Name/deployment")
  (a very common pattern for orchestrators, UIs, and inter-deployment calls) no longer finds the
  deployment, because both the flow name and — if the original deployment name used characters not
  permitted in a k8s object name, e.g. underscores — the deployment name have changed.
• Forced name mangling. Prefect deployment names allow underscores; k8s object names (RFC 1123) do
  not. So a deployment named generic_thematic_search cannot be reproduced at all — it becomes
  generic-thematic-search — even though Prefect itself is perfectly happy with the underscore.

This is the main blocker for adopting the operator for an existing fleet of deployments: migration
isn't transparent, and the (flow_name, deployment_name) identity that downstream code depends on
can't be preserved.

Note this is purely a metadata/identity problem, not an execution one — at run time the worker
re-imports the entrypoint and reconstructs the decorated Flow, so runs execute correctly regardless;
it's the deployment/flow records (and therefore by-name lookups) that diverge.

Proposed solution

Add two optional fields to PrefectDeploymentConfiguration (api/v1/prefectdeployment_types.go):

// FlowName overrides the flow name the deployment is registered under. Defaults to the
// entrypoint function token when unset.
// +optional
FlowName *string `json:"flowName,omitempty"`

// Name overrides the Prefect deployment name. Defaults to the CR's metadata.name when unset.
// Useful when the desired Prefect name isn't a valid Kubernetes object name (e.g. underscores).
// +optional
Name *string `json:"name,omitempty"`

Then honor them in convert.go:

• GetFlowIDFromDeployment: flowName := after → use *FlowName if set.
• ConvertToDeploymentSpec: Name: k8sDeployment.Name → use *Name if set.

Both default to current behavior when unset, so this is fully backward-compatible with no CRD
migration. POST /deployments/ already upserts on (flow_id, name), so idempotency is preserved as
long as the overridden values are used consistently.

Alternatives considered

• create-prefect-deployment.py already introspects the flow via load_flow_from_entrypoint, but
  it only uses that to populate parameterOpenApiSchema — it doesn't (and the CRD can't) set the flow
  name, so it doesn't address this.
• Renaming flows to match the entrypoint function name — not viable for existing fleets; breaks the
  decorator-name contract and every by-name consumer.

Happy to open a PR if this direction sounds good.

[jthetzel]

I also have this issue. We have existing flows with kebab-case flow names:

@flow(name="flow-foo")
def flow_foo():
    print("bar")

And we have another flow that uses prefect.deployment.run_deployment to run the deployment with the name flow-foo/deployment-foo. The prefect.yaml template suggests Prefect supports specifying the flow_name: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/deployments/templates/prefect.yaml#L19 . Thanks for the explanation, @sghegoiu-rokos .

[mitchnielsen]

Thank you both! Taking a look here.