Does anyone have Slack alerts set up with dbt tests on Prefect? Is it possible to send an alert with dbt source tests that are warning or failed?

You could do something like this:

@task(trigger=any_failed)
def send_slack_alert_on_failure(output):
    SlackTask(message=output).run()

this post provides a full example

Ok this could potentially work- However, with the test command I get passing and warning tests- is there a pay to only send the dbt output messages that include "warn"?

You would somehow need to filter by the log message content. It would be possible to do with something like Grafana - you could send all your dbt logs to some log aggregation service and then configure alerts based on message content Alternative: This could be a bit hacky, but you could do something like this in your flow:

@task(trigger=any_failed)
def send_slack_alert_on_failure(output):
    for line in output:
        if "WARN" in line:
            SlackTask(message=output).run()

getting this error when adding that code, what's the best way to debug this?

Beginning health checks...
Traceback (most recent call last):
  File "/opt/prefect/healthcheck.py", line 152, in <module>
    flows = cloudpickle_deserialization_check(flow_file_paths)
  File "/opt/prefect/healthcheck.py", line 44, in cloudpickle_deserialization_check
    flows.append(cloudpickle.loads(flow_bytes))
TypeError: code() takes at most 15 arguments (16 given)

try adding checkpoint=False to the task decorator on this task, otherwise the task run result may not get serialized properly

hmm still getting the error

even though it actually looks like an error from Docker storage - you could disable the health checks on your Docker storage using

ignore_healthchecks=True

, but the health checks are for a reason, looks like something in your flow code cannot be serialized with cloudpickle. If you need help debugging this, can you share your flow code?

I did get an error about the Python versions however the code works with that error when I don’t include that task so I figured it was something else

I see, perhaps worth matching the Python versions and trying again? doesn't hurt for sure

Yeah this is just in dev, in prod I will set a dependency between the syncs and dbt tests

you would set it in your base image

I don't currently specify any python version there

retry_on

is a recently added argument to the Task class, which makes me believe there is still some version mismatch between your registration and your prod runtime environments. Looks like your registration/build environment uses a more recent version of Prefect where this new argument exists, while your runtime/production environment is on an older version of Prefect and doesn't yet have this argument

retry_on

how do I update these environments so that they match?

I believe if you e.g. want the most recent Prefect version in both environments, you could do:

pip install prefect --upgrade

in both

locally and where else?

your registration environment (probably your development machine or CI system) and the Dockerfile if you explicitly set the image version there. If you don't use Dockerfile, Prefect will try to match the Prefect version with your environment

For some reason I'm not receiving the messages with "WARN"- will printing out the line here print to the Prefect logs?

@task(trigger=all_finished, checkpoint=False)
def send_slack_alert_on_test(output):
    for line in output:
        if "WARN" in line:
            print(line)
            SlackTask(message=line).run()

will printing out the line here print to the Prefect logs?

A really good question! I think it should work, as long as you add

log_stdout=True

to your task decorator (which you did) 🤔 Alternatively, you could do the same using logger which is safer:

logger = prefect.context.get("logger")
for line in output:
        if "WARN" in line:
            <http://logger.info|logger.info>(line)

hmmm yeah nothing is printing out

honestly no, I don't. Can you try with a simple flow and a simple dbt DAG and try to tackle it more step by step? does this work without filtering for WARN messages only?

no it doesn't

Yes there is! you would need to do that instead:

logger = prefect.context.get("logger")
if "WARN" in output:
    <http://logger.info|logger.info>(output) # or put it into Slack task

when I just print the entire output I only get that last message

that's true, I thought you have that already

cool thanks!