Welcome to the new DOCS.vastdata.com. We've added new menus and a version selector, plus new persona and resource pages to help customers get started. Looking for a previous article? Use the docs search.

Press CTRL+K to open search

Orchestrators (Airflow and similar)

Prev Next

Applies to VAST Cluster 5.5.

If a workflow in Airflow, Dagster, Prefect or Argo Workflows needs a DataEngine pipeline to process something as one of its steps, the step has to start the pipeline, wait for the run to finish, and fail if the run failed. This page shows how to do that with the DataEngine REST API and S3, from a plain Python script that any orchestrator can run as a task. It also shows how to deploy a pipeline over REST, for a workflow that sets up its own pipeline.

This isn't a built-in DataEngine integration in 5.5. There's no Airflow provider or operator for DataEngine, and the DataEngine API has no call that runs a pipeline on demand: pipelines run when their triggers fire. So the task fires the pipeline the way any upload would, by writing an object that its element trigger watches, and then looks for the run in DataEngine's telemetry.

The outputs come from a lab cluster running VAST Cluster 5.5.0-SP6, with Python 3.9, requests 2.32.5 and boto3 1.42.97 on the machine that ran the script. The pipeline ingest has an element trigger for new objects under incoming/ in the bucket docs-src, linked to a function hello. Your names, GUIDs, addresses and times will differ.

How a task runs a pipeline

  1. Check that the pipeline is Ready, with GET /api/dataengine/pipelines?name=<pipeline>.
  2. Fire it by writing an object, with a key unique to this run, under the trigger's bucket and prefix.
  3. Find the run. Writing the object returns no run or event ID, and the telemetry API can't search by object key. So the function logs one known line per object, including the key, and the task searches DataEngine's logs for exactly that line. The log record has the run's trace ID.
  4. Wait on the trace with GET /api/dataengine/telemetries/traces?trace_id=<id> until its status is OK or ERROR.
  5. Report: print the run's logs, and on ERROR its error logs, and exit non-zero so the orchestrator marks the task failed.

If your function writes an output (an object or a table row keyed by the input), the task can instead wait for that output with the orchestrator's own sensor, such as Airflow's S3KeySensor; then it needs only S3. The telemetry approach below works for any function, as long as it logs the line the task looks for.

Authenticate

The DataEngine API is served by VMS under https://<VMS address>/api/dataengine/, and its telemetry under https://<VMS address>/api/dataengine/telemetries/. Basic authentication names your tenant in a header, and you get a JWT for the tenant you name in the token request. In VAST Cluster 5.5.0-SP6, these worked:

Method How Result
Basic -u <user>:<password> plus the header X-Tenant-Name: <tenant> Works on the whole DataEngine API. Without X-Tenant-Name, the cluster answers 403 Invalid username/password.
JWT POST /api/token/<tenant> with the user name and password, then Authorization: Bearer <access token> Works on the whole DataEngine API. POST /api/token/ without the tenant fails for a tenant user.
VMS API token Authorization: Api-Token <token> Works on the telemetry endpoints only. Pipelines, functions, triggers and deploy answer 422 Invalid Api-Token.

So a task that only watches runs can use an API token, but a task that checks or deploys pipelines needs Basic or a JWT. Keep the user name and password in your orchestrator's secret store (for example an Airflow Connection) and pass them to the task as environment variables. The KB's authentication page describes the three methods.

The commands below use these variables. -k skips the certificate check, for a lab VMS with a self-signed certificate; leave it out if your VMS certificate is trusted:

export VMS_URL="https://<VMS address>"
export VAST_TENANT="<tenant>"
export VAST_USERNAME="<user>"
export VAST_PASSWORD="<password>"

Basic authentication:

curl -sk -u "$VAST_USERNAME:$VAST_PASSWORD" -H "X-Tenant-Name: $VAST_TENANT" "$VMS_URL/api/dataengine/pipelines?name=thumbnails" | jq ".data[] | {name, guid, status, reason, last_revision_number}"

{
  "name": "thumbnails",
  "guid": "c7f63133-3438-4cae-af6a-91291041f1b6",
  "status": "Ready",
  "reason": "Pipeline is ready and active.",
  "last_revision_number": 5
}

Without the tenant header:

curl -sk -w "\nHTTP %{http_code}\n" -u "$VAST_USERNAME:$VAST_PASSWORD" "$VMS_URL/api/dataengine/pipelines?name=thumbnails"

{"detail":"Invalid username/password."}
HTTP 403

A JWT. The response holds an access token and a refresh token:

curl -sk -X POST "$VMS_URL/api/token/$VAST_TENANT" -H "Content-Type: application/json" -d "{\"username\": \"$VAST_USERNAME\", \"password\": \"$VAST_PASSWORD\"}" | jq "keys"

[
  "access",
  "csrftoken",
  "refresh"
]

export TOKEN=$(curl -sk -X POST "$VMS_URL/api/token/$VAST_TENANT" -H "Content-Type: application/json" -d "{\"username\": \"$VAST_USERNAME\", \"password\": \"$VAST_PASSWORD\"}" | jq -r .access)
curl -sk -H "Authorization: Bearer $TOKEN" "$VMS_URL/api/dataengine/pipelines?name=thumbnails" | jq -c ".data[] | {name, status}"

{"name":"thumbnails","status":"Ready"}

A VMS API token, in API_TOKEN, on a pipeline and then on the telemetry:

curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $API_TOKEN" "$VMS_URL/api/dataengine/pipelines?name=thumbnails"

{"detail":"{\"authorization\": \"Value error, Invalid Api-Token\"}"}
HTTP 422

curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Api-Token $API_TOKEN" "$VMS_URL/api/dataengine/telemetries/traces?pipeline_name=thumbnails&limit=1"

{"pagination":{"next_cursor":"A|1791462329840025|8d3144dcbc26b05e21a4c6e59554e4fc", … },"data":[{"trace_id":"8d3144dcbc26b05e21a4c6e59554e4fc","start_time":1791462329840025,"end_time":1791462329842063,"status":"OK","trace_type":"init", … }]}
HTTP 200

The access token is short-lived, so get a new one at the start of each task, as the script below does, rather than storing it.

Make the function log a known line

The task finds the run by its log line, so the function must log one line per object that the task can predict. The hello function used here logs the whole event, then this line:

ctx.logger.info("hello: new object bucket=%s key=%s", props["bucket"], props["object_key"])

The telemetry API's message filter matches the whole message, not part of it. A search for the key alone finds nothing:

curl -sk -G -H "Authorization: Bearer $TOKEN" "$VMS_URL/api/dataengine/telemetries/logs" --data-urlencode "pipeline_name=ingest" --data-urlencode "message=rest-1.txt" | jq -c "{n: (.data | length)}"

{"n":0}

The whole line finds the record, and its trace ID:

curl -sk -G -H "Authorization: Bearer $TOKEN" "$VMS_URL/api/dataengine/telemetries/logs" --data-urlencode "pipeline_name=ingest" --data-urlencode "message=hello: new object bucket=docs-src key=incoming/rest-1.txt" | jq -r ".data[] | [.timestamp, .trace_id, .message] | @tsv"

1791467184316628	000050050000000e78c91f7f3863496c	hello: new object bucket=docs-src key=incoming/rest-1.txt

The task script

This script does the five steps. It authenticates with a JWT for the DataEngine API, and writes the object with boto3 using the S3 keys in AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY. Save it as orchestrate.py:

"""What an Airflow (or Dagster, Prefect, Argo) task does to run a DataEngine pipeline once:
check the pipeline is Ready, fire it by writing an object, wait for the run's trace, report.

Env: VMS_URL, VAST_TENANT, VAST_USERNAME, VAST_PASSWORD (REST); S3_ENDPOINT, AWS_ACCESS_KEY_ID,
AWS_SECRET_ACCESS_KEY (object write). usage: python orchestrate.py <pipeline> <bucket> <prefix> <run id>
"""
import os
import sys
import time

import boto3
import requests
import urllib3

urllib3.disable_warnings()  # test cluster with a self-signed certificate; pass a CA bundle in production
pipeline, bucket, prefix, run_id = sys.argv[1:5]
vms, tenant = os.environ["VMS_URL"], os.environ["VAST_TENANT"]

# 1. Authenticate: a JWT from /api/token/<tenant>, sent as a Bearer token
r = requests.post(f"{vms}/api/token/{tenant}", verify=False, json={
    "username": os.environ["VAST_USERNAME"], "password": os.environ["VAST_PASSWORD"]})
r.raise_for_status()
api = requests.Session()
api.verify = False
api.headers["Authorization"] = f"Bearer {r.json()['access']}"
de = f"{vms}/api/dataengine"

# 2. The pipeline must be Ready
p = api.get(f"{de}/pipelines", params={"name": pipeline}).json()["data"][0]
print(f"pipeline {p['name']}: {p['status']} ({p['reason']})")
if p["status"] != "Ready":
    sys.exit(1)

# 3. Fire: write a run-unique object under the trigger's prefix
key = f"{prefix}{run_id}.json"
started_us = time.time_ns() // 1000
s3 = boto3.client("s3", endpoint_url=os.environ["S3_ENDPOINT"], region_name="us-east-1")
s3.put_object(Bucket=bucket, Key=key, Body=f'{{"run_id": "{run_id}"}}'.encode())
print(f"wrote s3://{bucket}/{key}")

# 4. Wait: the function logs one known line per object; an exact-match log query finds it and its trace id
marker = f"hello: new object bucket={bucket} key={key}"
deadline = time.time() + 120
trace_id = None
while time.time() < deadline and trace_id is None:
    logs = api.get(f"{de}/telemetries/logs", params={
        "pipeline_name": pipeline, "message": marker, "time_after": started_us}).json()["data"]
    if logs:
        trace_id = logs[0]["trace_id"]
    else:
        time.sleep(2)
if trace_id is None:
    sys.exit(f"no run seen for {key} after 120 s")
trace = None
while time.time() < deadline:
    data = api.get(f"{de}/telemetries/traces", params={"trace_id": trace_id}).json()["data"]
    if data and data[0]["status"] in ("OK", "ERROR"):
        trace = data[0]
        break
    time.sleep(2)
print(f"trace {trace_id}: status {trace['status']}, event {trace['event']['event_id']}, "
      f"function {trace['resource']['function_name']}, "
      f"{(trace['start_time'] - started_us) / 1e6:.2f} s after the write, "
      f"ran {(trace['end_time'] - trace['start_time']) / 1000:.1f} ms")

# 5. Report: the run's user logs; on ERROR, its error logs; non-zero exit fails the task
if trace["status"] == "ERROR":
    errors = api.get(f"{de}/telemetries/logs", params={"trace_id": trace_id, "min_severity": "ERROR"}).json()["data"]
    for e in errors:
        print("  ERROR", e["message"])
    sys.exit(1)
for log in api.get(f"{de}/telemetries/logs", params={"trace_id": trace_id, "scope": "user"}).json()["data"]:
    print("  log", log["message"][:100])

The script turns off certificate checks because the lab VMS has a self-signed certificate. Against your cluster, set verify to your CA bundle's path instead, and drop urllib3.disable_warnings(). Change marker to the line your function logs. Telemetry times are in microseconds since the epoch, so the script records the time of the write in microseconds and asks only for logs after it.

Run it with the variables from authenticate, the S3 endpoint and keys, and a run ID; an orchestrator would pass its own run ID:

export S3_ENDPOINT="http://<S3 address>"
export AWS_ACCESS_KEY_ID="<access key>"
export AWS_SECRET_ACCESS_KEY="<secret key>"
python -W ignore orchestrate.py ingest docs-src incoming/ run-2026-10-08T1423; echo "exit $?"

pipeline ingest: Ready (Pipeline is ready and active.)
wrote s3://docs-src/incoming/run-2026-10-08T1423.json
trace 000050050000001b78c91f7f3863496c: status OK, event 87982405058587, function hello, 0.80 s after the write, ran 2.0 ms
  log CE attributes: "{'source': 'vastdata.com:incoming-uploads.19c9c0a7-e131-49d0-8aaa-1e34cc880721', 'id
  log CE data (dict): {'Records': [{'eventVersion': '2.2', 'eventSource': 'vast:s3', 'awsRegion': 'loop111
  log VastEvent properties: {"broker": "vast-broker-docs-kafka", "bucket": "docs-src", "extensions": {"dat
  log hello: new object bucket=docs-src key=incoming/run-2026-10-08T1423.json
exit 0

On the lab cluster, with the function's instance already running, the run started 0.8 seconds after the write (0.9 seconds in a second test). That's an observation on a small test cluster, not a guarantee: allow for more when an instance has to start, for example on the first event after a deploy, and set the task's timeout (120 seconds in the script) to suit your pipeline.

If the link to the function has retries, a failing event is retried before it's given up, and then goes to the dead-letter topic: see dead-letter topic. The script reports the first run it finds for the object.

An Airflow task

In Airflow, the script runs as one task. This DAG is a sketch: it wasn't run for this page. It passes the credentials from Airflow Connections as environment variables, and uses the DAG run's timestamp as the run ID:

from datetime import datetime

from airflow import DAG
from airflow.hooks.base import BaseHook
from airflow.operators.bash import BashOperator

vms = BaseHook.get_connection("vast_vms")     # host = https://<VMS address>, login, password, extra {"tenant": ...}
s3 = BaseHook.get_connection("vast_s3")       # login = access key, password = secret key, host = S3 endpoint

with DAG("ingest_with_dataengine", start_date=datetime(2026, 10, 1), schedule=None, catchup=False):
    BashOperator(
        task_id="run_ingest_pipeline",
        bash_command="python /opt/airflow/dags/orchestrate.py ingest docs-src incoming/ run-{{ ts_nodash }}",
        env={
            "VMS_URL": vms.host,
            "VAST_TENANT": vms.extra_dejson["tenant"],
            "VAST_USERNAME": vms.login,
            "VAST_PASSWORD": vms.password,
            "S3_ENDPOINT": s3.host,
            "AWS_ACCESS_KEY_ID": s3.login,
            "AWS_SECRET_ACCESS_KEY": s3.password,
        },
    )

Dagster, Prefect and Argo Workflows take the same shape: one op, task or container step that runs the script and fails on a non-zero exit.

Deploy a pipeline over REST

A workflow can also create and deploy the pipeline it uses. Most teams do this from CI with vastde instead (see CI/CD and automation), but the REST calls are short. The body of each call is the same object you'd write in YAML for vastde, as JSON; see pipeline YAML.

Create the element trigger. Save this as incoming-trigger.json, with your bucket:

{
  "name": "incoming-uploads",
  "description": "objects under incoming/ in docs-src",
  "config": {
    "source_bucket_name": "docs-src",
    "events": ["ObjectCreated:*"],
    "name_filters": {"prefixes": ["incoming/"]}
  }
}

curl -sk -u "$VAST_USERNAME:$VAST_PASSWORD" -H "X-Tenant-Name: $VAST_TENANT" -X POST "$VMS_URL/api/dataengine/triggers/element" -H "Content-Type: application/json" -d @incoming-trigger.json | jq "{name, guid, vrn, status, type, config}"

{
  "name": "incoming-uploads",
  "guid": "19c9c0a7-e131-49d0-8aaa-1e34cc880721",
  "vrn": "vast:dataengine:triggers:incoming-uploads",
  "status": "Ready",
  "type": "Element",
  "config": {
    "events": [
      "ObjectCreated:*"
    ],
    "source_bucket_name": "docs-src",
    "tag_filters": null,
    "name_filters": {
      "prefixes": [
        "incoming/"
      ],
      "suffixes": null
    }
  }
}

Create the pipeline. Save this as ingest-pipeline.json, with your compute cluster and namespace:

{
  "name": "ingest",
  "description": "created and deployed over REST",
  "kubernetes_cluster_vrn": "vast:dataengine:kubernetes-clusters:test-cluster",
  "namespace": "dataengine-ns-da47bae0-aefd-4e85-a5a9-2020d6a2cb91",
  "manifest": {
    "triggers": [{"name": "uploads", "vrn": "vast:dataengine:triggers:incoming-uploads"}],
    "function_deployments": [{"name": "hello", "function_vrn": "vast:dataengine:functions:hello"}],
    "links": [{"source": ["uploads"], "destination": ["hello"]}]
  }
}

curl -sk -u "$VAST_USERNAME:$VAST_PASSWORD" -H "X-Tenant-Name: $VAST_TENANT" -X POST "$VMS_URL/api/dataengine/pipelines" -H "Content-Type: application/json" -d @ingest-pipeline.json | jq "{name, guid, status, reason, last_revision_number}"

{
  "name": "ingest",
  "guid": "bc7b6df0-4dbc-4c69-94b7-61153607b1c3",
  "status": null,
  "reason": null,
  "last_revision_number": 1
}

Creating a pipeline doesn't deploy it: its status stays empty until you do. The deploy call is POST /api/dataengine/pipeline/<guid>/deploy, with pipeline in the singular. The plural path answers 404:

curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Bearer $TOKEN" -X POST "$VMS_URL/api/dataengine/pipelines/bc7b6df0-4dbc-4c69-94b7-61153607b1c3/deploy"

{"detail":"Not Found"}
HTTP 404

curl -sk -w "\nHTTP %{http_code}\n" -H "Authorization: Bearer $TOKEN" -X POST "$VMS_URL/api/dataengine/pipeline/bc7b6df0-4dbc-4c69-94b7-61153607b1c3/deploy"; curl -sk -H "Authorization: Bearer $TOKEN" "$VMS_URL/api/dataengine/pipelines?name=ingest" | jq -c ".data[] | {status, reason}"


HTTP 200
{"status":"InProgress","reason":null}

Then poll until the status is Ready or Error. Decide on status, not on reason: reason can still hold the text from an earlier state while the status is InProgress.

for i in $(seq 1 30); do s=$(curl -sk -H "Authorization: Bearer $TOKEN" "$VMS_URL/api/dataengine/pipelines?name=ingest" | jq -c ".data[0] | {status, reason}"); echo "$(date -u +%T) $s"; case "$s" in *\"Ready\"*|*\"Error\"*) break;; esac; sleep 5; done

13:45:56 {"status":"InProgress","reason":null}
13:46:01 {"status":"InProgress","reason":null}
13:46:07 {"status":"InProgress","reason":null}
13:46:13 {"status":"Ready","reason":"Pipeline is ready and active."}

The DataEngine API has two specs, which the cluster serves to authenticated users: https://<VMS address>/api/dataengine/openapi.json (functions, triggers, pipelines and deploy) and https://<VMS address>/api/dataengine/telemetries/openapi.json (logs, traces and metrics). The API reference is at kb.vastdata.com/api-5-5/apidocs, including deploying a pipeline.

See also

Related articles