Prefect

Step 1 of 6: Choose the deployment

Which environment this is: production, staging.

One key per deployment: a production and a staging install each get their own, and everything they send is kept apart by it.

It is fixed when the key is issued and cannot be changed afterwards.

This is also how it is listed: the tool and its deployment, such as Airflow in Production.

Step 2 of 6: Install the plugin

Add convalesce-emit-prefect where your flows run.

Two functions, attached to your flows and tasks with Prefect's own on_completion/on_failure hooks: no block to save, nothing to register with Prefect itself.

source /path/to/prefect-venv/bin/activate
pip install "convalesce-emit-prefect==0.2.1"

Step 3 of 6: Issue the key and set it

Point flow runs at Convalesce.

Set CONVALESCE_ENDPOINT and CONVALESCE_INGEST_KEY in the environment your flow runs execute in. On a work pool that is the job's environment.

With each run it sends the code of the flow and of each task, the SQL each task ran (the statement text, never the values bound to it) and the values each was called with, with secrets masked. CONVALESCE_SEND_SOURCE, CONVALESCE_SQL_CAPTURE and CONVALESCE_SEND_ARGUMENTS each turn one of those off when set to false.

Everything is sent to api.convalesce.io over HTTPS, port 443, from each flow run. That call out has to be allowed there; nothing has to be opened inbound.

export CONVALESCE_ENDPOINT="https://api.convalesce.io/openapi"
export CONVALESCE_INGEST_KEY="<your-ingest-key>"

The process that runs the flow, such as flow.serve(); restart it afterwards.

Step 4 of 6: Attach the hooks

emit_flow_run on flows, emit_task_run on tasks.

emit_flow_run on flows and emit_task_run on tasks, in both on_completion and on_failure. On flows, add on_crashed and on_cancellation too: a flow that crashes or is cancelled never reaches on_failure, and without them its run never shows as finished.

Import convalesce_emit_prefect before the flow opens its database connections, so the SQL each task runs is seen.

Delivery problems never raise into your flow. Check the log of the process that runs the flow for convalesce: warnings; a missing key is logged there.

from prefect import flow, task
from convalesce_emit_prefect import emit_flow_run, emit_task_run


@task(on_completion=[emit_task_run], on_failure=[emit_task_run])
def transform(data):
    return data.split(" ")


@flow(
    on_completion=[emit_flow_run],
    on_failure=[emit_flow_run],
    on_crashed=[emit_flow_run],
    on_cancellation=[emit_flow_run],
)
def etl():
    transform("This is data")

Step 5 of 6: Turn on retries, if you want them (optional)

Optional: let Prefect run a flow again once somebody approves it.

When Convalesce proposes rerunning a failed flow run and somebody approves it here, your own Prefect carries it out. Nothing reaches into Prefect from outside: a small flow asks Convalesce every minute what has been approved for this deployment, with the key you have already set, and schedules those flow runs again through the Prefect API.

The flow below is all it takes.

A flow run can be run again when it belongs to a deployment, since that is what a worker picks up.

from prefect import flow


@flow(name="convalesce-retries", log_prints=True)
def convalesce_retries():
    from convalesce_emit_prefect.retry import run_pending_retries

    print(run_pending_retries())


if __name__ == "__main__":
    convalesce_retries.serve(name="every-minute", cron="* * * * *")

Run it with python convalesce_retries.py and leave it running, or deploy the flow to a work pool on the same schedule, in the same environment as your other flows: it uses the two Convalesce variables already set there and Prefect's own address. Each run asks Convalesce what has been approved for this deployment and schedules those flow runs again.

Step 6 of 6: Check it reports

Run something and watch the first event arrive.

Run the check below where the tool runs, or start a run there. Then press the button: it shows Connected within a minute of the key first reaching Convalesce.