convalesce-emit-dagster

Forward Dagster run status to Convalesce, unchanged.

Install

pip install convalesce-emit-dagster

Wire it up

The package exports a sensor body rather than a decorated sensor, because Dagster's decorator signatures move between versions and this package does not depend on Dagster. Apply the decorator in your own definitions:

from dagster import DagsterRunStatus, Definitions, run_status_sensor
from convalesce_emit_dagster import convalesce_sensor


@run_status_sensor(run_status=DagsterRunStatus.SUCCESS)
def convalesce_on_success(context):
    convalesce_sensor(context)


@run_status_sensor(run_status=DagsterRunStatus.FAILURE)
def convalesce_on_failure(context):
    convalesce_sensor(context)


@run_status_sensor(run_status=DagsterRunStatus.CANCELED)
def convalesce_on_canceled(context):
    convalesce_sensor(context)


defs = Definitions(
    sensors=[convalesce_on_success, convalesce_on_failure, convalesce_on_canceled]
)

Each fires with the run, the event and the sensor name, as Dagster produced them. A cancelled run is recorded as skipped. Set CONVALESCE_INGEST_KEY where the daemon runs.

Links to the Dagster UI

Set CONVALESCE_DAGSTER_URL where your sensor runs to the address you open the Dagster UI at, such as https://dagster.example.com, or https://my-org.dagster.cloud on Dagster+. Each job, op, run and step in Convalesce then links to its page in Dagster. On Dagster+ the link goes through the deployment the run ran in, a branch deployment included.

What each step ran

With the run, the sensor sends one entry per executed step:

• the source of the op or asset function behind it, read from your definitions;
• the SQL statements it ran through SQLAlchemy or a database driver, as text, with the dialect and database and without bound values;
• the configuration it was resolved with.

The statements and the configuration are noted in the process that runs the step and written to the run's event log as one engine event per step, which the sensor reads back. This starts when your definitions import convalesce_emit_dagster, on the multiprocess and the in-process executor alike. Secrets are masked before anything is written or sent.

Each is on by default and has its own switch, set where your steps and your sensor run: CONVALESCE_SEND_SOURCE, CONVALESCE_SQL_CAPTURE and CONVALESCE_SEND_ARGUMENTS (false turns one off).

What an asset's author attached

Metadata on a materialisation, an observation, an asset check and a Failure is sent in two parts:

• the entries that describe a table, as they are: row counts, the table name and URI, the column schema and column lineage, the code version and the SQL an asset ran;
• every other entry whose value is a number, a flag or a timestamp, under its own name: MetadataValue.int, .float, .bool and .timestamp, or the bare value. Up to 50 per event, with names up to 128 characters.

Text, markdown, JSON, tables, paths and links are replaced by a count of what was withheld. Set CONVALESCE_DAGSTER_SEND_METADATA=false where your sensor runs to send the first part only.

The path an IO manager records is sent when it is the URI of a place in an object store, such as s3://my-bucket/orders, which names a dataset. A path on a local disk is counted with what was withheld. Set CONVALESCE_DAGSTER_SEND_LOCAL_PATHS=true where your sensor runs to send local paths too.

What identifies the deployment

On Dagster+ the sensor sends the deployment name, whether it is a branch deployment, the code location, the commit and branch, the repository and its URL, and the pull request number and status, as the agent sets them in DAGSTER_CLOUD_* variables. The commit's author and message stay where they are.

A run's tags are sent, yours and Dagster's dagster/ ones. Of the .dagster/ tags Dagster keeps for itself, the scheduled execution time and the repository are sent. The login of whoever launched the run, the agent's id and label, the process id, and the agent's environment variables stay where they are.

Declare what an op reads and writes

An asset names its own table. For an op whose datasets appear nowhere in its code or metadata, pass them to the sensor as lineage, keyed by op name:

LINEAGE = {
    "load_orders": {
        "inputs": ["s3://my-bucket/exports/orders"],
        "outputs": [
            "urn:cvl:dataset:(urn:cvl:dataPlatform:snowflake,my_db.my_schema.orders,PROD)"
        ],
    },
}


@run_status_sensor(run_status=DagsterRunStatus.SUCCESS)
def convalesce_on_success(context):
    convalesce_sensor(context, lineage=LINEAGE)

A dataset is named by its urn, as in the convalesce.inputs and convalesce.outputs metadata of an op's inputs and outputs, or by the URI of a path in an object store. lineage may also be a function of the sensor's context that returns such a mapping, for a declaration that depends on the run.

Settings

Set CONVALESCE_SEND_SETTINGS=true where your steps run and each step's entry also carries the environment variables its process had. It is off by default.

The settings are read in the process that runs the step, so a step on the multiprocess executor or behind a step launcher reports its own environment. They travel in the same engine event as the rest of what the step ran, already in the form they are sent in. Set CONVALESCE_INGEST_KEY or CONVALESCE_FINGERPRINT_KEY where your steps run, so there is a key to hash with there.

A setting whose name or value reads as a credential, and any value longer than 300 characters, is sent as a fingerprint: a keyed hash, sixteen hex characters long, made in your process. It says that the value changed between two runs and nothing of what the value is. Every other setting is sent as its value.

The hash is keyed with CONVALESCE_FINGERPRINT_KEY where you set one, and with a key derived from your ingest key otherwise. The fingerprint key is only ever read in your process and is never sent. keyed_by identifies the key the hashes were made under, so hashes are compared only between runs that used the same one.

Set CONVALESCE_SETTINGS_SKIP to a comma-separated list of names to keep those settings out in either form, such as CONVALESCE_SETTINGS_SKIP=HOSTNAME,INTERNAL_REGION.

Each step's entry then holds:

{
  "settings": {
    "items": [
      {"kind": "environment", "name": "API_KEY", "fingerprint": "5d1c0a9e7b3f2468"},
      {"kind": "environment", "name": "WAREHOUSE", "value": "analytics"}
    ],
    "keyed_by": "9f3b6c1d2e4a5b70"
  }
}

Output of a failed step

A step that fails also says what it logged last. Its entry in steps carries output_tail: the last lines the step logged through context.log, get_dagster_logger() or a library's own logger, as they were written, followed by the error it failed with, each cause's message under the three innermost frames of its stack. A step that succeeds sends none of it, and neither does one that failed on one try and succeeded on a later one.

At most 60 lines and 4096 bytes are sent, the end of the output being the part that is kept. A password in a URI, the value of a name such as password, token or api_key, a JSON Web Token and a PEM block are masked first, and a masked or shortened tail is declared in the event's excluded. What a step writes to stdout and stderr is Dagster's compute log, and is not part of the tail.

Like the rest of a step's entry, the tail is noted in the step's process and written to the run's event log, where the sensor reads it.

Set CONVALESCE_SEND_OUTPUT=false where your steps run to note none of it.

The entry then holds:

{
  "step_key": "load_orders",
  "output_tail": "Running with dbt=1.8.7\ninvalid identifier 'SALES_AMOUNT'\nValueError: dbt build failed"
}

Supported

Dagster 1.7 and later, on Python 3.9 and later. Verified on real installs: see version-support.md.

Configure

Read from the environment: see convalesce-emit.