--- title: "Orchestrators (Airflow and similar)" slug: "dev-dataengine-integrations-orchestrators" description: "Learn how to integrate DataEngine pipelines with Airflow, Dagster, Prefect, or Argo Workflows using REST API and S3 for efficient data processing." tags: ["DataEngine"] updated: 2026-10-30T18:19:01Z published: 2026-10-30T18:19:01Z canonical: "docs.vastdata.com/dev-dataengine-integrations-orchestrators" --- > ## Documentation Index > Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt > Use this file to discover all available pages before exploring further. # Orchestrators (Airflow and similar) *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](https://kb.vastdata.com/api-5-5/apidocs/engine-api-authentication) 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://" export VAST_TENANT="" export VAST_USERNAME="" export VAST_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 """ 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/, 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](/develop/docs/dev-dataengine-integrations-orchestrators#authenticate), the S3 endpoint and keys, and a run ID; an orchestrator would pass its own run ID: ``` export S3_ENDPOINT="http://" export AWS_ACCESS_KEY_ID="" export AWS_SECRET_ACCESS_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](/develop/docs/dev-dataengine-integrations-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://, 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](/develop/docs/dev-dataengine-operations-ci-cd)), 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](/develop/docs/dev-dataengine-reference-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](https://kb.vastdata.com/api-5-5/apidocs), including [deploying a pipeline](https://kb.vastdata.com/api-5-5/apidocs/deploy-a-pipeline-to-kubernetes). ## See also - [CI/CD and automation](/develop/docs/dev-dataengine-operations-ci-cd): `vastde`, the REST API and `vastpy` in CI. - [Automation tools](/develop/docs/dev-database-reference-automation-tools): how the REST API, `vastpy`, Terraform and the admin MCP server compare. - [Observability](/develop/docs/dev-dataengine-guides-observability): the same logs and traces from `vastde`. - [Triggers](/develop/docs/dev-dataengine-guides-triggers#element-triggers): prefixes, suffixes and tag filters for the object the task writes. - [Security](/develop/docs/dev-dataengine-operations-security): who can call which DataEngine API. - [Integrations](/develop/docs/dev-dataengine-integrations-index)