This guide explains how to configure OpenLineage on an on-premise Airflow cluster so that ADOC can observe your DAG runs.
For Airflow versions later than 2.7, use the apache-airflow-providers-openlineage package.
For Airflow versions 2.7 and earlier, use the openlineage-airflow package.
In the airflow.cfg file located in $AIRFLOW_HOME, add an [openlineage] section and configure one of the following two authentication options.
This is the simplest option. Provide your ADOC access key and secret key directly as HTTP headers using the custom_headers configuration:
transport={"type": "http","url": "<host name of the ADOC tenant>","endpoint": "/torch-pipeline/api/v1/lineage","verify": false,"custom_headers": {"accessKey": "<access key>","secretKey": "<secret key>"}}
namespace="<openlineage namespace - choose an appropriate name>"
With this configuration, ADOC receives your access key and secret key through the accessKey and secretKey HTTP headers.
The token provider implementation and the access_key_secret_token_provider.py file described in Option 2 are not required when using this option.
If you prefer not to include your access key and secret key directly in the transport configuration, you can provide these credentials through an Airflow Connection instead. In this case, the transport configuration references a token provider rather than embedding credentials:
transport={"type": "http","url": "<host name of the ADOC tenant>", "endpoint": "/torch-pipeline/api/v1/lineage", "auth": {"type": "tokenproviders.access_key_secret_token_provider.AccessKeySecretKeyTokenProvider"}}
namespace='<openlineage namespace - choose an appropriate name>'
Log in to the Airflow UI.
Navigate to Admin → Connections.
Click Add (➕) to create a new connection.
Fill in the following details:
Field | Value | Details |
|---|
Connection ID | acceldata_connection
| — |
Connection Type | HTTP
| — |
Host | <Host name of your ADOC tenant>
| Use the same host value as defined in the transport configuration |
Login | <Your ADOC Access Key>
| — |
Password | <Your ADOC Secret Key>
| — |
Click Save to create the connection.
Once saved, Airflow securely manages your credentials, so you can omit them from the main transport configuration.
In the $AIRFLOW_HOME/plugins folder, create a directory called tokenproviders, and inside it create a file named access_key_secret_token_provider.py with the following contents:
# Copyright 2018-2023 contributors to the OpenLineage project
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import json
import logging
import os
from datetime import datetime, timedelta
from typing import Any
import requests
from airflow.configuration import conf
from airflow.hooks.base import BaseHook
from airflow.models import Variable
from dateutil import parser
from openlineage.client.transport.http import TokenProvider
log = logging.getLogger(__name__)
os.environ["no_proxy"] = "*"
CLIENT_ID = "acceldata-app"
TOKEN_PROVIDER_PATH = "/admin/api/onboarding/token-exchange?grant_type=api_keys"
ACCELDATA_LINEAGE_URL = "acceldata_lineage_url"
ACCELDATA_LINEAGE_ENDPOINT = "acceldata_lineage_endpoint"
ACCELDATA_ACCESS_KEY = "acceldata_access_key"
ACCELDATA_SECRET_KEY = "acceldata_secret_key"
ACCELDATA_BEARER_TOKEN = "acceldata_bearer_token"
ACCELDATA_EXPIRES_AT = "acceldata_expires_at"
class AccessKeySecretKeyTokenProvider(TokenProvider):
def __init__(self, config: dict[str, str]) -> None:
super().__init__(config)
self.access_key = config.get("access_key")
self.secret_key = config.get("secret_key")
if self.is_config_loaded():
log.info("Config is already loaded. Skipping loading from the config.")
else:
ol_config = self._load_openlineage_config()
access_key, secret_key, credential_source, transport = self._resolve_credentials(ol_config)
if not access_key or not secret_key:
log.error(
"Credential resolution failed. Neither Airflow Connection nor "
"OpenLineage config provides valid access_key and secret_key."
)
return
self._finalize_and_persist(
transport=transport,
access_key=access_key,
secret_key=secret_key,
credential_source=credential_source,
)
def _load_openlineage_config(self) -> dict[str, str]:
return self.load_config()
def _resolve_credentials(self, config):
transport, access_key_from_config, secret_key_from_config = self.extract_transport_and_auth(config)
# Try the Airflow Connection first
try:
conn = BaseHook.get_connection("acceldata_connection")
if conn.login and conn.password:
return conn.login, conn.password, "airflow_connection", transport
except Exception as e:
log.info("Airflow Connection 'acceldata_connection' not available. Falling back to OpenLineage config. Details=%s", e)
# Fall back to the OpenLineage config
if access_key_from_config and secret_key_from_config:
return access_key_from_config, secret_key_from_config, "openlineage_config", transport
return None, None, None, None
def _finalize_and_persist(self, *, transport, access_key: str, secret_key: str, credential_source: str):
self.access_key = access_key
self.secret_key = secret_key
self.persist_config_to_airflow_variables(transport, self.access_key, self.secret_key)
@staticmethod
def is_config_loaded():
from airflow.models import Variable
acceldata_lineage_url = Variable.get(ACCELDATA_LINEAGE_URL, None)
acceldata_lineage_endpoint = Variable.get(ACCELDATA_LINEAGE_ENDPOINT, None)
acceldata_secret_key = Variable.get(ACCELDATA_SECRET_KEY, None)
acceldata_access_key = Variable.get(ACCELDATA_ACCESS_KEY, None)
return all([acceldata_lineage_url, acceldata_lineage_endpoint, acceldata_secret_key, acceldata_access_key])
@staticmethod
def persist_url(transport):
if "url" in transport:
Variable.set(ACCELDATA_LINEAGE_URL, transport["url"])
else:
log.error("Missing required OpenLineage configuration: transport.url not found")
@staticmethod
def persist_endpoint(transport):
if "endpoint" in transport:
Variable.set(ACCELDATA_LINEAGE_ENDPOINT, transport["endpoint"])
else:
log.error("Missing required OpenLineage configuration: transport.endpoint not found")
@staticmethod
def persist_access_key(access_key: str | None):
if access_key:
Variable.set(ACCELDATA_ACCESS_KEY, access_key)
@staticmethod
def persist_secret_key(secret_key: str | None):
if secret_key:
Variable.set(ACCELDATA_SECRET_KEY, secret_key)
def load_config(self) -> dict[str, Any]:
try:
openlineage_config = conf.getsection('openlineage')
if not isinstance(openlineage_config, dict):
log.error("Invalid OpenLineage configuration type. Expected dict but got %s", type(openlineage_config).__name__)
return {}
return openlineage_config
except Exception as e:
log.error("Failed to read [openlineage] configuration from airflow.cfg. Error=%s", e)
return {}
@staticmethod
def get_bearer_token(token):
return f"Bearer {token}"
@staticmethod
def _update_token_to_cache(token: str, expires_in: int):
expiration = datetime.now() + timedelta(seconds=expires_in)
Variable.set(ACCELDATA_BEARER_TOKEN, token)
Variable.set(ACCELDATA_EXPIRES_AT, expiration)
@staticmethod
def _get_token_from_cache():
cached_token = Variable.get(ACCELDATA_BEARER_TOKEN, None)
expires_at = Variable.get(ACCELDATA_EXPIRES_AT, None)
return cached_token, expires_at
@staticmethod
def validate_token(cached_token, expires_at):
if expires_at is None:
return False
expires_at_date = parser.parse(expires_at)
return cached_token is not None and expires_at_date > datetime.now()
def _fetch_token_from_admin_central(self):
try:
headers = {'Content-Type': 'application/json'}
access_key = Variable.get(ACCELDATA_ACCESS_KEY, None)
secret_key = Variable.get(ACCELDATA_SECRET_KEY, None)
data = {'clientId': CLIENT_ID, 'secretKey': secret_key, 'accessKey': access_key}
token_provider_base_url = Variable.get(ACCELDATA_LINEAGE_URL, None)
if token_provider_base_url is not None:
token_provider_url = token_provider_base_url + TOKEN_PROVIDER_PATH
response = requests.post(token_provider_url, json=data, headers=headers)
if response.status_code == 200:
token_data = response.json()
token = token_data.get('access_token')
expires_in = token_data.get('expires_in') - 60
return token, expires_in
else:
log.error("Failed to fetch bearer token from Admin Central (status_code=%s)", response.status_code)
return None, None
else:
log.warning("OpenLineage backend URL is not configured in Airflow Variables.")
except Exception as e:
log.error("Exception occurred while fetching bearer token from Admin Central. Error=%s", e)
def refresh_token(self):
token, expires_in = self._fetch_token_from_admin_central()
if token:
self._update_token_to_cache(token, expires_in)
else:
log.error("Bearer token refresh failed")
@staticmethod
def extract_transport_and_auth(config):
if "transport" not in config:
log.error("Missing required OpenLineage configuration: transport")
return None, None, None
try:
transport = json.loads(config.get("transport"))
except Exception as e:
log.error("Failed to parse OpenLineage transport config. Error=%s", e)
return None, None, None
auth = transport.get("auth", {})
return transport, auth.get("access_key"), auth.get("secret_key")
def persist_config_to_airflow_variables(self, transport, access_key, secret_key):
self.persist_url(transport)
self.persist_endpoint(transport)
self.persist_access_key(access_key)
self.persist_secret_key(secret_key)
def get_bearer(self):
cached_token, expires_at = self._get_token_from_cache()
if self.validate_token(cached_token, expires_at):
return self.get_bearer_token(cached_token)
self.refresh_token()
cached_token, expires_at = self._get_token_from_cache()
return self.get_bearer_token(cached_token)
This class defines the token provider referenced in the auth section of the transport configuration above.
You can add further configuration — such as enabling selective DAG processing — by extending the OpenLineage configuration in airflow.cfg, following the standard [section] -> key = value format.
Restart all Airflow components to apply the plugin and configuration changes.
With this setup, any DAGs placed in the $AIRFLOW_HOME/dags folder are automatically monitored on the ADOC platform through the OpenLineage integration.
