https://prefect.io logo
Title
a

Apoorva Desai

03/14/2022, 9:40 PM
Hello, I am trying to add slack notifications for a prefect flow using this: https://docs.prefect.io/core/advanced_tutorials/slack-notifications.html#installation-instructions I have a slack webhook URL that I've saved as a prefect secret. I am importing this:
from prefect.utilities.notifications import slack_notifier
and my tasks now look like this :
task(log_stdout=True, state_handlers=[slack_notifier])

install_snowflake_task = ShellTask(helper_script="pip install boto3 \
    snowflake-connector-python[pandas] \
    snowflake-ingest && pip install PyJWT==1.7.1", shell="bash", stream_output=True, return_all=True, state_handlers=[slack_notifier])

install_dbt_task = ShellTask(helper_script="pip install dbt==0.18.0", \
                             shell="bash", stream_output=True, return_all=True, state_handlers=[slack_notifier])
My flow looks like
with Flow("name-of-flow", state_handlers=[slack_notifier]) as flow:
The flow runs successfully but I see no notifications on the slack channel that I have authorized for this. What am I doing wrong?
a

Anna Geller

03/14/2022, 9:42 PM
This page is quite old tbh 😄 it's much easier to create a Slack webhook, store it as Secret in your Prefect Cloud account and use it in the
SlackTask
LMK if you need an example
a

Apoorva Desai

03/14/2022, 9:46 PM
Would love an example!
👍 1
Also, do you have a link for instructions please?
a

Anna Geller

03/14/2022, 9:47 PM
yup, let me write it first 😄
❤️ 1
LMK if you have any questions about it 🙂
a

Apoorva Desai

03/14/2022, 10:02 PM
Thank you!
👍 1
So I followed those instructions and saved my custom secret. I keep getting this error from SlackTask though:
Error during execution of task: ValueError('Local Secret "PREFECT__CONTEXT__SECRETS__SLACK_WEBHOOK_URL_PREFECT" was not found.')
I can confirm that this secret exists...
This is how I'm using it:
slack = SlackTask(webhook_secret="PREFECT__CONTEXT__SECRETS__SLACK_WEBHOOK_URL_PREFECT")
a

Anna Geller

03/14/2022, 11:52 PM
Run this first:
export PREFECT__CLOUD__USE_LOCAL_SECRETS=false
this will ensure that Prefect doesn't look for local Secrets but rather those stored in the Cloud backend
1
a

Apoorva Desai

03/15/2022, 12:09 AM
This should be in the prefect flow.py file and should be run each time?
That didn't work either
a

Anna Geller

03/15/2022, 1:05 AM
No, you need to set it only once and this will be stored in your ~/.prefect/config.toml
You can also set it manually directly there
[cloud]
use_local_secrets = false
a

Apoorva Desai

03/15/2022, 1:14 AM
So I am running prefect on a k8 cluster and I'm using sealedsecrets for prefect secrets. Wouldn't this make it a local secret?
a

Anna Geller

03/15/2022, 1:14 AM
if you need to set it on your agent:
prefect agent kubernetes start --env PREFECT__CLOUD__USE_LOCAL_SECRETS=false
or set it as part of env variables in your yaml file
spec:
      containers:
      - args:
        - prefect agent kubernetes start
        command:
        - /bin/bash
        - -c
        env:
        - name: PREFECT__CLOUD__USE_LOCAL_SECRETS
          value: false
1
a

Apoorva Desai

03/15/2022, 4:11 PM
@Anna Geller Thank you for your help yesterday. I have Prefect now posting messages to slack. I ended up having to initialize the SlackTask like this:
slack = SlackTask()
and passed in a
webhook_url
instead of secret in the flow like this:
slack(message="Hi from a flow!", webhook_url=SLACK_WEBHOOK_URL_PREFECT)
. This works and I see the message on the chosen channel. However, I still get a failure from SlackTask at some point in the flow with this message:
Task 'SlackTask': Exception encountered during task execution!
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 863, in get_task_run_state
    logger=self.logger,
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/executors.py", line 454, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/tasks.py", line 445, in method
    return run_method(self, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/tasks/notifications/slack_task.py", line 61, in run
    webhook_url = webhook_url or cast(str, Secret(webhook_secret).get())
  File "/usr/local/lib/python3.7/site-packages/prefect/client/secrets.py", line 142, in get
    ) from None
ValueError: Local Secret "SLACK_WEBHOOK_URL" was not found.
Also, the state handler function you shared fails too:
Exception raised while calling state handlers: AttributeError("'Context' object has no attribute 'task_name'")
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/cloud/flow_runner.py", line 120, in call_runner_target_handlers
    old_state=old_state, new_state=new_state
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/flow_runner.py", line 116, in call_runner_target_handlers
    new_state = handler(self.flow, old_state, new_state) or new_state
  File "<string>", line 25, in post_to_slack_on_failure
AttributeError: 'Context' object has no attribute 'task_name'
Are there some imports I am missing? Maybe something for old_state, new_state?
Bumping this if anyone has a solution ^
k

Kevin Kho

03/15/2022, 7:57 PM
SlackTask does not take in
webhook_url
. It takes in a
webhook_secret
that is a string that is used to fetch the Secret with that name from Prefect Cloud. Because you used
webhook_url
, it looked for the default secret name which is
SLACK_WEBHOOK_URL
:upvote: 1
a

Anna Geller

03/15/2022, 7:57 PM
can you share your flow and how do you call your flow? do you run your flow just locally?
This shouldn't happen if you set the config to use cloud Secrets rather than the local ones as we discussed before
ValueError: Local Secret "SLACK_WEBHOOK_URL" was not found
it's actually puzzling because flow runs triggered by an agent should pull Cloud secrets by default - what is your prefect version? Can you share the output of your "prefect diagnostics" if you still see some issues?
but first and most important question: can you confirm that you are on Prefect Cloud? otherwise we need to start over here if you are on Server 😂
a

Apoorva Desai

03/15/2022, 8:11 PM
@Lon Nix just confirmed that we're on Prefect Server 😬
a

Anna Geller

03/15/2022, 8:12 PM
upsss
in that case follow this topic: https://discourse.prefect.io/t/how-to-set-storage-secrets-e-g-github-access-token-on-server/70 just instead of
PREFECT__CONTEXT__SECRETS__GITHUB_ACCESS_TOKEN
you need
PREFECT__CONTEXT__SECRETS__SLACK_WEBHOOK_URL_PREFECT
a

Apoorva Desai

03/15/2022, 8:37 PM
I am already doing that, this is my flow:
I see the "Hi from a flow" message on my slack channel of choice but don't see anything from the state handlers
a

Anna Geller

03/15/2022, 9:24 PM
Kevin already told you that you should set
webhook_secret
rather than
webhook_url
here:
SlackTask(message=msg, webhook_url=SLACK_WEBHOOK_URL_PREFECT).run()
also, the state handler will fire only when your task will fail. The way you set this up at the moment it's super unlikely to fail because it does nothing other than printing something to the console 😄
@task(log_stdout=True, state_handlers=[post_to_slack_on_failure])
def print_output(output):
    logger = prefect.context.get("logger")
    for l in output:
        <http://logger.info|logger.info>(l)
so you should change it to:
SlackTask(message=msg, webhook_secret=SLACK_WEBHOOK_URL_PREFECT).run()
provided you set the Secret as we discussed e.g.:
export PREFECT__CONTEXT__SECRETS__SLACK_WEBHOOK_URL_PREFECT=xxx
but if you set it to the default name, you don't have to set it explicitly - I would actually recommend that:
export PREFECT__CONTEXT__SECRETS__SLACK_WEBHOOK_URL=xxx
then you can just do:
SlackTask(message=msg).run()
l

Lon Nix

03/15/2022, 9:28 PM
Thanks @Anna Geller. We found we were doing things one way in one file and another way in another file so we're adjusting from our side to make sure all the changes you've suggested got implemented properly across the board. Hopefully that takes care of our issues and we'll force a failure somewhere to test that state handler
🙌 1
a

Apoorva Desai

03/15/2022, 9:33 PM
Yeah, thanks @Anna Geller 🙂
👍 1