https://prefect.io
Join Slack
Hello all, I am working on setting up Slack Notifications with Prefect and I am running into some is...
# ask-community
m
Hello all, I am working on setting up Slack Notifications with Prefect and I am running into some issues the following code does not work but if I replace the request.post(my_secrect) call with the Slack webhook address directly it prints the is_failed() message as expected. What am I doing wrong with the configuration of the Secrets?
Copy code
def post_to_slack(task, old_state, new_state):
    my_secret = PrefectSecret("SLACK_WEBHOOK_URL").run()
    if new_state.is_retrying():
        msg = "Task {0} failed and is retrying at {1}".format(task, new_state.start_time)

        # replace URL with your Slack webhook URL
        <http://requests.post|requests.post>(my_secret, json={"text": msg})
    elif new_state.is_failed():
        msg = "Task {0} failed".format(task)

     # replace URL with your Slack webhook URL
        <http://requests.post|requests.post>(my_secret, json={"text": msg})

    return new_state
I specifically receive this message if I leave the code as it is above:
Copy code
Exception raised while calling state handlers: ClientError('400 Client Error: Bad Request for url: <http://host.docker.internal:4200/graphql>\n\nThe following error messages were provided by the GraphQL server:\n\n    GRAPHQL_VALIDATION_FAILED: Cannot query field "secret_value" on type "Query".\n\nThe GraphQL query was:\n\n    query($name: String!) {\n                secret_value(name: $name)\n    }\n\nThe passed variables were:\n\n    {"name": "SLACK_WEBHOOK_URL"}\n')
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/client/secrets.py", line 137, in get
    value = secrets[self.name]
KeyError: 'SLACK_WEBHOOK_URL'
Thank you in advance!
z
Hi @Matthew Blau! Looks like you’re trying to use secrets with Prefect Server — to do that you’ll need to use local secrets in the config/env instead. See https://docs.prefect.io/core/idioms/secrets.html#authentication-and-secrets-in-prefect
m
Hi @Zanie I have also swapped out PrefectSecrets for Secrets("SLACK_WEBHOOK_URL").get() and still receive a KeyError. I have the URL set in the config.toml under [context.secrets].
@Zanie just confirmed that the error still exists and I still get the KeyError with the code set to:
Copy code
def post_to_slack(task, old_state, new_state):
    my_secret = Secret("SLACK_WEBHOOK_URL").get()
    if new_state.is_retrying():
        msg = "Task {0} failed and is retrying at {1}".format(task, new_state.start_time)

        # replace URL with your Slack webhook URL
        <http://requests.post|requests.post>(my_secret, json={"text": msg})
    elif new_state.is_failed():
        msg = "Task {0} failed".format(task)

     # replace URL with your Slack webhook URL
        <http://requests.post|requests.post>(my_secret, json={"text": msg})

    return new_state
Am I missing something with how the secrets need to be handled?
I am at a complete loss as to what is failing here; I have determined that the requests.post() call is wanting a string so when I pass in the secret it is unsure of what to do with it, however when I cast it to a string in the assignment call, it still complains about a KeyError of SLACK_WEBHOOK_URL. No matter what I pass in as a secret it fails, despite the secret existing in my config.toml. Running from the interactive shell I can assign
Copy code
s = Secret("SLACK_WEBHOOK_URL") and then call s.get()
and it parses and prints out the proper value. So obviously the key exists and prefect can parse it. I am confused as to why it is unable to parse it in my post_to_slack() method
z
Sorry about the delay -- have your run 
prefect backend server
? Can you give me the output of 
prefect diagnostics
The code should not be reaching this exception, e.g.
Copy code
try:
            value = secrets[self.name]
        except KeyError:
            if prefect.config.backend != "cloud":
                raise ValueError(
                    'Local Secret "{}" was not found.'.format(self.name)
                ) from None
m
prefect diagnostics shows
Copy code
{
  "config_overrides": {
    "context": {
      "secrets": false
    }
  },
  "env_vars": [],
  "system_information": {
    "platform": "Linux-5.4.0-65-generic-x86_64-with-glibc2.29",
    "prefect_backend": "server",
    "prefect_version": "0.14.6",
    "python_version": "3.8.5"
  }
}
config.toml is
Copy code
[context.secrets]

SLACK_WEBHOOK_URL="url_here"
z
Is your agent running on the same machine?
m
@Zanie yes. I am using docker agent, everything is local to me currently.
z
I do not think secrets are passed from your local context into the docker agent -- they're explicitly disabled in the agent's env (https://github.com/PrefectHQ/prefect/blob/master/src/prefect/agent/docker/agent.py#L538)
Can you try a 
DockerRun
config with an environment variable?
Copy code
PREFECT__CONTEXT__SECRETS__SLACK_WEBHOOK_URL="foo"
m
@Zanie how would I be able to pass it to the docker agent? Currently what I have for the flow is:
Copy code
with Flow("example",
            storage = Docker(dockerfile="/home/mblau/projects/experian-integration/Dockerfile", ignore_healthchecks= False,
            
)) as flow:
    result = main()
what would you like for me to adjust here?
z
m
@Zanie passing it in via the -e flag when starting the docker agent works. Is this something that would need to be ran for every secret? Right now I have Prefect build the container with the provided Dockerfile.
z
I'm not sure what you mean. You could pull all the secrets from the config programmatically and insert them into the 
env
of a 
DockerRun
or generate the command to run the agent. The secrets aren't automatically passed through from a local context to the agent.
m
@Zanie right now, my understanding of how the Secrets work, is that I need to have the secrets in my config.toml in order to be able to use them in my flows. I have turned on use_local_secrets = true as well in the config file and without passing the Slack webhook url to the docker agent upon starting the agent I get the KeyError. I have the flow use DockerStorage, provide it the Dockerfile, and then I run
Copy code
python3 integration.py
in order to have the flow be registered with the UI. It works and errors expectedly, however, the Slack Notification does not fire off the message of "Task: test slack failed" without that runtime arg being passed to the DockerAgent
z
The secret is not passed into your container unless you do so explicitly so the flow does not have it. Only parts of your config are passed into the container by the agent.
I've opened an issue for the misleading error https://github.com/PrefectHQ/prefect/issues/4051 -- I do not think we should automatically pull secrets from the local config and pass them into the container but you are welcome to open an issue for that if you want that to happen automatically.
m
@Zanie could you provide an example of what you mean for passing it to a DockerRun? I feel that I am not understanding exactly what you are referring to.
z
Copy code
In [1]: from prefect import Flow
fr
In [2]: from prefect.run_configs import DockerRun

In [3]: with Flow("ex") as flow:
   ...:     pass
   ...: 

In [4]: from prefect import config

In [6]: flow.run_config = DockerRun(env={f"PREFECT__CONTEXT__SECRETS__{k}": v for k, v in config.context.secrets.items()})

In [7]: flow.run_config.env
Out[7]: {'PREFECT__CONTEXT__SECRETS__fooo': 'bar'}
There's also an example in the docs I linked before
m
@Zanie Thank you for your help yesterday! Prefect is a huge thing to learn and my company is super excited to roll this out for our workflow orchestration and this channel has been a big help with the process
z
Thanks @Matthew Blau 🙂
m
I do have one question for you, @Zanie, if you don't mind. Where can I find information what I am able to use in the state_handler I wrote for Slack notifications? I am wanting to have it output information like this photo. Is this just something to be investigated on the Slack docs side of things or Prefects?
z
There should be a good bit of information in the context https://docs.prefect.io/api/latest/utilities/context.html#context-2
It also receives the task object and old/new states that have information attached to them.
m
@Zanie so it looks like I can pass in any dict values to this? Basically am trying to capture and display more useful information than just "Some reference tasks failed" Basically something like this is more what I am wanting to output to the Slack notification. With prefect.context I would be able to grab that info easily and post it to Slack, yes? That is my understanding of things, at least
z
The context contains information provided by Prefect, I wouldn’t try to insert data into it at runtime. If you’re trying to capture information about a failed task from an exception, that is stored in the 
new_state.result
variable. I would highly recommend putting a 
breakpoint()
in your state handler and examining what is available to you in the case you’re interested in.
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by