Triggerer Service Management
Relevant source files
The following files were used as context for generating this wiki page:
The Triggerer Service Management in the Meltano Airflow extension facilitates the execution and lifecycle of the Airflow Triggerer process within a Meltano project environment. This service is part of the broader Airflow utility extension, which wraps the standard Apache Airflow CLI to provide integrated management of orchestration components.
The extension utilizes a passthrough mechanism to allow Meltano to invoke the airflow triggerer command directly while ensuring the necessary environment configurations, such as AIRFLOW_HOME and database initialization, are correctly applied. This ensures that the Triggerer, responsible for running deferred tasks (async operators), operates within the context of the Meltano project's specific Airflow configuration.
Sources: README.md:126, airflow_ext/wrapper.py:34-54
Service Invocation and Lifecycle
The Triggerer service is managed through the airflow_invoker and the primary extension entry point. It is typically started as a background process during the deployment or local development phase of a Meltano project.
Invocation Flow
The following diagram illustrates how a command to start the Triggerer flows through the extension:
The extension ensures that before the Triggerer service starts, the environment is initialized, and the database is ready.
Sources: airflow_ext/pass_through.py:12-20, airflow_ext/wrapper.py:56-64, README.md:126
Key Service Commands
Management of the Triggerer and related services is handled via the following standard commands:
| Command | Action | Description |
|---|---|---|
meltano invoke airflow:initialize | Initialization | Seeds the database and creates airflow.cfg. |
meltano invoke airflow triggerer & | Service Start | Starts the Airflow Triggerer process in the background. |
airflow_invoker | Passthrough | The script used to pass commands directly to the underlying airflow binary. |
Sources: README.md:113-126, pyproject.toml:40-41
Environment and Configuration
The Triggerer service relies on specific environment variables and configuration files managed by the Airflow class. These settings ensure that logs, database connections, and configurations are isolated to the Meltano project.
Critical Environment Variables
The following variables are essential for the Triggerer service to function correctly within the extension:
- AIRFLOW_HOME: Defines the directory where Airflow stores its configuration and logs. Defaults to
$MELTANO_PROJECT_ROOT/orchestrate/airflow. - AIRFLOW_CONFIG: The path to the
airflow.cfgfile, typically located within theAIRFLOW_HOME. - SLUGIFY_USES_TEXT_UNIDECODE: Hardcoded to
yesby the wrapper to ensure Airflow is installable without certain GPL dependencies.
Sources: airflow_ext/wrapper.py:39-54, README.md:92-105
Configuration Generation
Before the Triggerer is invoked, the extension checks for and generates a default configuration if one does not exist. This is handled by the _create_config method, which captures the default Airflow configuration via airflow config list --defaults.
Sources: airflow_ext/wrapper.py:113-134
Implementation Details
The Triggerer service management is implemented across several key files that handle the CLI mapping and process execution.
Passthrough Mechanism
The airflow_invoker script (defined in pyproject.toml) points to airflow_ext.pass_through:pass_through_cli. This function initializes the Airflow extension class and passes all command-line arguments (e.g., triggerer) directly to the airflow binary using an Invoker.
def pass_through_cli() -> None:
"""Pass through CLI entry point."""
pass_through_logging_config()
ext = Airflow()
ext.pass_through_invoker(
structlog.getLogger("airflow_invoker"),
*sys.argv[1:] if len(sys.argv) > 1 else [],
)
Sources: airflow_ext/pass_through.py:12-20, pyproject.toml:41
Execution Logic
The execution logic is encapsulated in the Airflow class within wrapper.py.
The Airflow class manages the lifecycle by calling pre_invoke (which triggers _create_config and _initdb) before any invoke or pass_through_invoker call. This ensures the Triggerer always has an up-to-date database schema.
Sources: airflow_ext/wrapper.py:34-111
Summary
Triggerer Service Management in this extension provides a seamless way to run Airflow's asynchronous task processor within a Meltano environment. By wrapping the standard Airflow CLI and managing initialization automatically, it ensures that the Triggerer is correctly configured with project-specific paths and database settings, allowing for background execution alongside other Airflow components like the scheduler and API server.
Sources: README.md:113-130, airflow_ext/wrapper.py:56-64