Prefect: Allow Cloud Hooks to be set in Flow.register()

[mhmcdonald]

Current behavior

Please describe how the feature works today

In order to set cloud hooks, you either have to do this through the Cloud UI or with the graphql api.

Proposed behavior

Please describe your proposed change to the current behavior

Ideally, I would maintain a version control of my cloud hooks (email/slack/webhook etc, recipients, and states) in the flow definition files. Then when I deploy my flows, these cloud hooks would be set when I call Flow.register().

Example

Please give an example of how the enhancement would be useful
My company is using prefect in a self service fashion (multiple teams write their own flows). Right now, I can't rely on each flow author to use the graph ql api or log into cloud to setup cloud hooks. Adding this to Flow.register() would allow me to automate the setup/maintenance of cloud hooks.

[joshmeek]

Thanks for the idea @mhmcdonald this is a great enhancement and shouldn't be too difficult since it's only a simple graphql mutation away. Any thoughts on that the .register should accept? I'm thinking a dictionary which matches the options you could send to the create_cloud_hook mutation.

mutation {
  create_cloud_hook(input: {type: EMAIL, name: "Example", version_group_id: "abc", states: ["Running"], config: {to: "[email protected]"}}) { # <-- this part
    id
  }
}

[mhmcdonald]

that should do it! I don't have anything else to add. Thanks @joshmeek

[mhmcdonald]

@joshmeek one update, I think the .register should accept a list of dictionaries, in case you want to register multiple cloud hooks for a single flow (slack + webhook or whatever).

I'd be interested in trying to working on this myself and contributing. However, the thing that's not quite clear to me is how I'd get the version_group_id if it's not provided in .register (it's an optional arg). When I deploy my flows I never specify this value. If it's a brand new flow being registered for the first time, I wouldn't yet have the version_group_id until after it's registered, correct?

.register() returns the flow_id. This value is different from version_group_id. If you could provide either the flow_id or the version_group_id I think I'd be able to get this to work. But that would require an update to the graphql api, correct?

[joshmeek]

@mhmcdonald I was thinking about this the other day but hadn't officially started any work on it yet so I'm glad you're interested in tackling it! As it stands right now the version_group_id would have to be queried for using the flow id returned from the create_flow_from_compressed_string mutation that is called during registration. Then that version group id could be used to call additional mutations to create the cloud hooks (this would only happen if any cloud hooks were provided to the .register() call).

I do like the idea of also returning the version_group_id from the create_flow_from_compressed_string mutation cc @zdhughes for some thoughts on this

[zdhughes]

Heya @joshmeek, not sure how I missed that CC.

We can definitely look into adding additional fields to the payload returned by register_flow / create_flow! As mentioned, we'd currently need to take the flow id we get at registration time and query for the version group ID.

Thankfully the pattern will remain the same, just with an eventually obsolete step:

• call the create_flow mutation
• query the API with something like this (which can be cut if the API is updated)

query {
  flow(where: {id: {_eq: "xyz"}}) {
    version_group_id
  }
}

• create the cloud hooks with the retrieved version group ID!

[mhmcdonald]

So, I've implemented the cloud hook creation on my end when I register my flows. One thing I've observed during my testing is that there is nothing enforcing uniqueness on cloud hook creation. I can create the exact same cloud hook with the same name, type, config etc, many times for a single flow.

On my end, to get around this, when I create the cloud hook, I first check if the cloud hook already exists (based on the name of the cloud hook).

cloud_hook_query = { "query": { with_args("cloud_hook", {"where": {"version_group_id": {"_eq": version_group_id}, "name": {"_eq": "api-webhook"}}}): { "name" } } }

I'm beginning to think that if the create_cloud_hook mutation does not enforce uniqueness based on the name of the cloud hook, adding this to flow.register() is not going to give people the result they'd expect. You wouldn't want to be slacked/emailed/etc n times where n is the number of times you registered the flow.

Am I thinking about this correctly @zdhughes and @joshmeek? Would you consider enforcing cloud hook uniqueness by name for a given flow?

[zdhughes]

Hi @mhmcdonald, I can confirm that the API does not enforce uniqueness of cloud hooks based on name, and I'm not sure we want to potentially limit users in that manner. Looking at the original issue though, I think we have a clean way forward.

Since cloud hooks are associated with version groups, users could create a cloud hook once when registering the first flow of a version group, which is a solid use case for creating a cloud hook at registration time. Because cloud hooks follow flows within a version group, that means the user wouldn't need to create a cloud hook each time they registered, avoiding the duplicate cloud hook pattern. Does this sound like it might fit your pattern?

[jlowin]

I would like to push back on this idea a bit - I don't think flow.register() is a natural place to add additional things like Cloud Hooks, as they are not part of flow registration (and are only possible after a flow has been registered)

However I think the motivation of creating hooks at creation time is a good one. As an alternative, I would suggest that we expose Cloud Hook management via new prefect.Client methods. That way you could still include them as part of a deploy script:

# deploy_script.py
with Flow("my flow") as flow:
    ...

flow_id = flow.register()
client.create_cloud_hook(flow_id, ...)

This atomic separation is much more natural to me; throwing everything into flow.register() is going to result in great complexity. For example, the flow being created but an error in cloud hook creation would lead to a very unsatisfactory user experience.

[mhmcdonald]

I agree @jlowin. The more time I've spent looking at this, the less sense it makes to add to the register method. I ended up adding it to my flow deployment script. I still think it would be a nice enhancement if flow.register() returned the version_group_id in addition to the flow_id, for convenience.

And yes! Adding the functionality to the client would be great. It's easier than having to work with the graphql mutation.

To follow up on @zdhughes response: there are probably some use cases that I'm not aware of re: cloud hook names being repeated for a flow. Based on how I deploy my flows, I'm not actively tracking whether this is the first time the flow is being registered or the 100th time. For that reason, my work around is to check if the cloud hook already exists, prior to trying to create it.

Right now, I have to execute 2 graphql queries and 1 mutation to deploy a single cloud hook.

1. query: get the version_group_id for the flow_id
2. query: check to see if the cloud hook already exists for this version_group_id
3. mutation to add the cloud_hook

This works fine for me, and I think it will for others. I'm will go ahead and close this issue, unless anyone would like for me to leave it open.