[Incompleted][SWIP-7][DISCUSSION] Support Apache Airflow monitoring

[songzhendong]

Motivation

Apache Airflow is an open-source workflow management platform primarily used for scheduling and monitoring workflows, It can be used to handle complex data pipelines and has been widely applied in the fields of data engineering and data science. Airflow allows users to write workflows, which are called DAGs (Directed Acyclic Graphs). Each DAG contains a series of tasks that can be executed in a specific sequence and dependency relationship, Due to its support for multitasking in complex scenarios, monitoring the health and operational status of Airflow is crucial, Through these metrics, it is possible to help analyze task health status, formulate optimization plans, and design risk prevention strategies.

Architecture Graph

There is no significant architecture-level change.

Proposed Changes

graph LR;
  AirflowOTEL("Airflow OTEL") --> OpenTelemetryCollector("OpenTelemetry Collector") --> SkyWalkingOTELReceiver("SkyWalking OTEL Receiver") -->  SkyWalkingMALEngine("SkyWalking MAL Engine")
 --> SkyWalkingUI("SkyWalking UI")

Loading

1. OpenTelemetry Collector fetches metrics from Airflow via OTEL and pushes metrics to SkyWalking OTEL
   Receiver via OpenTelemetry exporter.
2. The SkyWalking OAP Server parses the expression with MAL to filter/calculate/aggregate and store the results.
3. These metrics can be displayed via the SkyWalking UI, and the metrics can be customized for display on the UI dashboard.

Airflow Service Supported Metrics

Monitoring Panel	Unit	Metric Name	Description	Data Source
Airflow Job Started	count	<job_name>_start	Number of started job	OpenTelemetry from Airflow
Tasks Executable	count	scheduler.tasks.executable	Number of tasks that are ready for execution (set to queued) with respect to pool limits, DAG concurrency, executor state, and priority.	OpenTelemetry from Airflow
Tasks Cleared	count	scheduler.orphaned_tasks.cleared	Number of Orphaned tasks cleared by the Scheduler	OpenTelemetry from Airflow
Tasks Adopted	count	scheduler.orphaned_tasks.adopted	Number of Orphaned tasks adopted by the Scheduler	OpenTelemetry from Airflow
Queued Tasks	count	executor.queued_tasks	Number of queued tasks on executor	OpenTelemetry from Airflow
Pool Open Slots	count	executor.open_slots	Number of open slots on executor	OpenTelemetry from Airflow
Pool Queued Slots	count	pool.queued_slots	Number of queued slots in the pool. Metric with pool_name tagging.	OpenTelemetry from Airflow
Deferred Slots	count	pool.deferred_slots	Number of deferred slots in the pool. Metric with pool_name tagging.	OpenTelemetry from Airflow
Scheduler Heartbeat	rate	scheduler_heartbeat	Scheduler heartbeats	OpenTelemetry from Airflow
DAG File Queue Size	count	dag_processing.file_path_queue_size	Number of DAG files to be considered for the next scan	OpenTelemetry from Airflow
Dataset Updates	count	dataset.updates	Number of updated datasets	OpenTelemetry from Airflow

Airflow Instance Supported Metrics

Monitoring Panel	Unit	Metric Name	Description	Data Source
Airflow Job Started	count	<job_name>_start	Number of started job	OpenTelemetry from Airflow
Pool Open Slots	count	pool.open_slots	Number of open slots in the pool. Metric with pool_name tagging	OpenTelemetry from Airflow
Pool Deferred Slots	count	pool.deferred_slots	Number of deferred slots in the pool. Metric with pool_name tagging	OpenTelemetry from Airflow
Pool Running Slots	count	pool.running_slots	Number of running slots in the pool. Metric with pool_name tagging.	OpenTelemetry from Airflow
Triggerer Heartbeat	rate	triggerer_heartbeat	Number of open slots on executor	OpenTelemetry from Airflow
Triggers Main Thread	count	triggers.blocked_main_thread	Number of triggers that blocked the main thread (likely due to not being fully asynchronous)	OpenTelemetry from Airflow
Triggers Succeeded	count	pool.deferred_slots	Number of deferred slots in the pool. Metric with pool_name tagging.	OpenTelemetry from Airflow
triggers Failed	count	triggers.failed	Number of triggers that errored before they could fire an event	OpenTelemetry from Airflow
Tasks Executable	count	scheduler.tasks.executable	Number of tasks that are ready for execution (set to queued) with respect to pool limits, DAG concurrency, executor state, and priority.	OpenTelemetry from Airflow
Tasks Cleared	count	scheduler.orphaned_tasks.cleared	Number of Orphaned tasks cleared by the Scheduler	OpenTelemetry from Airflow
Tasks Adopted	count	scheduler.orphaned_tasks.adopted	Number of Orphaned tasks adopted by the Scheduler	OpenTelemetry from Airflow
Queued Tasks	count	executor.queued_tasks	Number of queued tasks on executor	OpenTelemetry from Airflow
Dataset Updates	count	dataset.updates	Number of updated datasets	OpenTelemetry from Airflow
Dataset Orphaned	count	dataset.orphaned	Number of datasets marked as orphans because they are no longer referenced in DAG schedule parameters or task outlets	OpenTelemetry from Airflow
Dataset Triggered Dagruns	count	dataset.triggered_dagruns	Number of DAG runs triggered by a dataset update	OpenTelemetry from Airflow

If the metrics exists for both service and its instances , then the number displayed on the service dashboard is the sum of all instances

Imported Dependencies libs and their licenses.

No new dependency.

Compatibility

no breaking changes.

General usage docs

[wu-sheng]

@wankai123 Please take a look.

[wu-sheng]

@songzhendong The graph should be drawn through mermaid, refer to https://raw.githubusercontent.com/apache/skywalking-showcase/main/docs/readme.md. It will show up through markdown rendering https://skywalking.apache.org/docs/skywalking-showcase/next/readme/

[songzhendong]

Thanks

[wankai123]

The proposal LGTM

If the metrics exists for both service and its instances , then the number displayed on the service dashboard is the sum of all instances

Notice:
If the OTEL collects the metrics from instance, then we couldn't sum them into service directly.
please refer to: #11982

[songzhendong]

Thanks

[wu-sheng]

The SWIP-7 is accepted, please go ahead to prepare your pull request.

Note, the SWIP should be included in your final pull request.

[songzhendong]

I will finish it later