description |
---|
Sending automated alerts to a Slack channel. |
{% hint style="warning" %} This is an older version of the ZenML documentation. To read and view the latest version please visit this up-to-date URL. {% endhint %}
The SlackAlerter
enables you to send messages to a dedicated Slack channel
directly from within your ZenML pipelines.
The slack
integration contains the following two standard steps:
- slack_alerter_post_step takes a string message or a custom Slack block, posts it to a Slack channel, and returns whether the operation was successful.
- slack_alerter_ask_step
also posts a message or a custom Slack block to a Slack channel, but waits for user feedback, and
only returns
True
if a user explicitly approved the operation from within Slack (e.g., by sending "approve" / "reject" to the bot in response).
Interacting with Slack from within your pipelines can be very useful in practice:
- The
slack_alerter_post_step
allows you to get notified immediately when failures happen (e.g., model performance degradation, data drift, ...), - The
slack_alerter_ask_step
allows you to integrate a human-in-the-loop into your pipelines before executing critical steps, such as deploying new models.
Before you can use the SlackAlerter
, you first need to install ZenML's slack
integration:
zenml integration install slack -y
{% hint style="info" %} See the Integrations page for more details on ZenML integrations and how to install and use them. {% endhint %}
In order to use the SlackAlerter
, you first need to have a Slack workspace set up with a channel that you want your
pipelines to post to.
Then, you need to create a Slack App with a bot in your workspace.
{% hint style="info" %}
Make sure to give your Slack bot chat:write
and chat:write.public
permissions in the OAuth & Permissions
tab
under Scopes
.
{% endhint %}
Next, you need to register a slack
alerter in ZenML and link it to the bot you just created. You can do this with the
following command:
zenml alerter register slack_alerter \
--flavor=slack \
--slack_token=<SLACK_TOKEN> \
--default_slack_channel_id=<SLACK_CHANNEL_ID>
Here is where you can find the required parameters:
<SLACK_CHANNEL_ID>
: Open your desired Slack channel in a browser, and copy out the last part of the URL starting withC....
.<SLACK_TOKEN>
: This is the Slack token of your bot. You can find it in the Slack app settings underOAuth & Permissions
. IMPORTANT: Please make sure that the token is theBot User OAuth Token
not theUser OAuth Token
.
After you have registered the slack_alerter
, you can add it to your stack like this:
zenml stack register ... -al slack_alerter
After you have a SlackAlerter
configured in your stack, you can directly import
the slack_alerter_post_step
and slack_alerter_ask_step
steps and use them in your pipelines.
Since these steps expect a string message as input (which needs to be the output of another step), you typically also need to define a dedicated formatter step that takes whatever data you want to communicate and generates the string message that the alerter should post.
As an example, adding slack_alerter_ask_step()
to your pipeline could look like this:
from zenml.integrations.slack.steps.slack_alerter_ask_step import slack_alerter_ask_step
from zenml import step, pipeline
@step
def my_formatter_step(artifact_to_be_communicated) -> str:
return f"Here is my artifact {artifact_to_be_communicated}!"
@pipeline
def my_pipeline(...):
...
artifact_to_be_communicated = ...
message = my_formatter_step(artifact_to_be_communicated)
approved = slack_alerter_ask_step(message)
... # Potentially have different behavior in subsequent steps if `approved`
if __name__ == "__main__":
my_pipeline()
An example of adding a custom Slack block as part of any alerter logic for your pipeline could look like this:
from typing import List, Dict
from zenml.integrations.slack.steps.slack_alerter_ask_step import slack_alerter_post_step
from zenml.integrations.slack.alerters.slack_alerter import SlackAlerterParameters
from zenml import step, pipeline
@step
def my_custom_block_step(block_message) -> List[Dict]:
my_custom_block = [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f":tada: {block_message}",
"emoji": true
}
}
]
return SlackAlerterParameters(blocks = my_custom_block)
@pipeline
def my_pipeline(...):
...
message_blocks = my_custom_block_step("my custom block!")
post_message = slack_alerter_post_step(params = message_blocks)
return post_message
if __name__ == "__main__":
my_pipeline()
For more information and a full list of configurable attributes of the Slack alerter, check out the API Docs .