Flyte webhooks

[pingsutw]

TL;DR

Add a new webhook processor capable of parsing the workflow event and sending the message to an external service, such as Slack.

1. Flyteadmin sends a workflow event to SNS
2. SNS push the event to SQS
3. FlyteAdmin webhook processor pulls the event from SQS
4. Parse the event, and then render the payload template ({{ name }} succeeded)
5. Send the event to external service by calling the webhook endpoint

Type

• Bug Fix
• Feature
• Plugin

Are all requirements met?

• Code completed
• Smoke tested
• Unit tests added
• Code documentation added
• Any pending items have an associated Issue

Complete description

webhook config

externalEvents:
  enable: true
  type: "aws"
  aws:
    region: "us-west-2"
  eventsPublisher:
    topicName: "arn:aws:sns:us-west-2:590375264460:webhook"
    eventTypes:
      - all
webhooknotifications:
  aws:
    region: "us-west-2"
  webhooks:
    name: slack
    url: https://hooks.slack.com/services/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
    payload: "{{ name }} succeeded"
    processor:
      queueName: webhook
      accountId: "590375264460"

Tracking Issue

flyteorg/flyte#2317

Follow-up issue

NA

[codecov]

Codecov Report

Merging #583 (fd83160) into master (d492640) will increase coverage by 1.10%.
The diff coverage is 39.80%.

❗ Current head fd83160 differs from pull request most recent head 38869de. Consider uploading reports for the commit 38869de to get more accurate results

@@            Coverage Diff             @@
##           master     #583      +/-   ##
==========================================
+ Coverage   58.68%   59.78%   +1.10%     
==========================================
  Files         171      171              
  Lines       16484    13444    -3040     
==========================================
- Hits         9674     8038    -1636     
+ Misses       5958     4552    -1406     
- Partials      852      854       +2     

Flag	Coverage Δ
unittests	?

Flags with carried forward coverage won't be shown. Click here to find out more.

Files Changed	Coverage Δ
pkg/rpc/adminservice/base.go	3.70% <0.00%> (+0.31%)	⬆️
pkg/runtime/application_config_provider.go	30.00% <0.00%> (-3.34%)	⬇️
pkg/async/webhook/factory.go	8.33% <8.33%> (ø)
pkg/async/webhook/implementations/slack_webhook.go	16.21% <16.21%> (ø)
pkg/manager/impl/execution_manager.go	73.06% <20.00%> (+2.48%)	⬆️
pkg/async/webhook/implementations/processer.go	54.41% <54.41%> (ø)
...ync/notifications/implementations/aws_processor.go	59.45% <72.22%> (-9.68%)	⬇️
...ync/notifications/implementations/gcp_processor.go	66.66% <85.71%> (-1.34%)	⬇️
pkg/async/notifications/email.go	100.00% <100.00%> (ø)
...g/async/webhook/implementations/webhook_metrics.go	100.00% <100.00%> (ø)

... and 153 files with indirect coverage changes

[fg91]

@pingsutw will this work without any managed AWS services like SNS, i.e. also on other cloud providers like GCP?

[pingsutw]

@fg91, we'll support GCP pub/sub. We'd like to add Redis or Kafka publisher for the on-prem cluster but in separate PR.

[fg91]

@fg91, we'll support GCP pub/sub. We'd like to add Redis or Kafka publisher for the on-prem cluster but in separate PR.

This is awesome, thanks 🙏

[fg91]

When using this new feature to send notifications to slack via webhooks (instead of via emails), how will the UX look like in flytekit in the end?

wacky_int_doubler_lp = LaunchPlan.get_or_create(
    notifications=[
        Slack(
            phases=[
                WorkflowExecutionPhase.SUCCEEDED,
            ],
            recipients_email=["[email protected]"],  # What will this be replaced with?
        ),
    ],
)

(Source)

Will there be the option to choose a specific webhook, e.g. via name (which in the end corresponds to a Slack channel if I understand correctly)?

[pingsutw]

Something like this.

lp = LaunchPlan.get_or_create(
    notifications=[
        Webhook(
            name=<webhook_name>,  # Should match the name in Flyte admin config map.
            phases=[
                WorkflowExecutionPhase.SUCCEEDED,
            ],
            payload=["{{ name succeeded }}"],
        ),
    ],
)

In the current implementation, the admin will call the webhook endpoint when the task is completed.

[fg91]

        Webhook(
            name=<webhook_name>,  # Should match the name in Flyte admin config map.

This sounds great 👍

[eapolinario]

Hi, we are moving all Flyte development to a monorepo. In order to help the transition period, we're moving open PRs to monorepo automatically and your PR was moved to flyteorg/flyte#4147. Notice that if there are any conflicts in the resulting PR they most likely happen due to the change in the import path of the flyte components.