1fanwang opened a new pull request, #71561: URL: https://github.com/apache/airflow/pull/71561
In HA deployments every scheduler runs `_emit_pool_metrics` on its own loop and samples the metadata DB a moment apart. Pool metrics are tagged only by `pool_name` (plus `team_name` under multi-team), so every scheduler publishes the *same* series and a gauge keeps whichever export landed last. `pool.open_slots` flaps between schedulers' samples and is frequently wrong. Tagging the gauge per scheduler would fix the collision but multiplies pool series by the scheduler count. A histogram instead folds every scheduler into one series per pool and retains every sample, so the true value stays recoverable — `min` for `open_slots`, `max` for the queued/running/deferred/scheduled counts. This emits a `stats.timing` alongside each existing gauge: same value, same tags, `_histogram` suffix. The gauges are unchanged, so existing scrapers keep working. An unbounded pool (`slots=-1`) reports `open=inf`, which is skipped rather than recorded — OpenTelemetry drops a non-finite observation but logs a warning on every scheduler emission, and StatsD would put a literal `inf|ms` on the wire. closes: https://github.com/apache/airflow/issues/66800 Replaces https://github.com/apache/airflow/pull/66810, which was approved by @xBis7 and @ferruzzi and then auto-closed as a stale draft. Both review points raised there are folded in here: the registry descriptions now state these are unit-less slot counts rather than durations, and the infinite-pool case is handled. Still open for a wider discussion (not this PR): `timing` is the only histogram primitive the stats API exposes, so a slot count rides the timer type. On OpenTelemetry that is the intended `Histogram`; on StatsD it renders as `|ms`. # Testing Done `_emit_pool_metrics` is exercised against a real `Pool` row, and the collision itself is reproduced through the real `SafeOtelLogger` and OpenTelemetry SDK. | # | Scenario | Result | |---|---|---| | 1 | Two schedulers report divergent `open_slots` for one pool | Gauge keeps last writer (wrong); histogram preserves both samples (correct) | | 2 | Gauge/histogram pair for all five slot states | Same value and tags | | 3 | Unbounded pool (`slots=-1`) | Gauge carries `inf`; `open_slots_histogram` not emitted | **1. The bug, through the real OTel logger.** Two `SafeOtelLogger` instances share one `InMemoryMetricReader`, standing in for two schedulers exporting to one backend: ```python from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import InMemoryMetricReader from airflow_shared.observability.metrics.otel_logger import SafeOtelLogger reader = InMemoryMetricReader() provider = MeterProvider(metric_readers=[reader]) scheduler_a = SafeOtelLogger(otel_provider=provider) scheduler_b = SafeOtelLogger(otel_provider=provider) tags = {"pool_name": "default_pool"} for scheduler, open_slots in ((scheduler_b, 126), (scheduler_a, 128)): scheduler.gauge("pool.open_slots", open_slots, tags=tags) scheduler.timing("pool.open_slots_histogram", open_slots, tags=tags) # an unbounded pool would report inf into the same histogram scheduler_a.timing("pool.queued_slots_histogram", 4, tags=tags) scheduler_a.timing("pool.queued_slots_histogram", float("inf"), tags=tags) for metric in reader.get_metrics_data().resource_metrics[0].scope_metrics[0].metrics: point = next(iter(metric.data.data_points)) if metric.name.endswith("_histogram"): print(f"{metric.name}: count={point.count} sum={point.sum} min={point.min} max={point.max}") else: print(f"{metric.name}: value={point.value}") ``` <details><summary>Raw output</summary> ``` $ uv run --project shared/observability python dev/repro_pool_histogram.py Record amount inf is not finite on Histogram airflow.pool.queued_slots_histogram, ignoring measurement. airflow.pool.open_slots: value=128 airflow.pool.open_slots_histogram: count=2 sum=254.0 min=126.0 max=128.0 airflow.pool.queued_slots_histogram: count=1 sum=4.0 min=4.0 max=4.0 ``` Scheduler B's `126` is gone from the gauge — the backend reports `128`. The histogram kept both, so `min` recovers the correct `126`. The `inf` sample also shows why the guard is worth having: the SDK discards the measurement but logs that line on every scheduler emission. StatsD has no such filter — the raw client puts it on the wire: ``` $ python -c "import statsd; ...; real.timing('pool.open_slots_histogram', float('inf'))" statsd wire payload: call(b'pool.open_slots_histogram:inf|ms', ('127.0.0.1', 8125)) ``` </details> **2. Regression tests, red then green.** Both new tests fail on unmodified `scheduler_job_runner.py`: <details><summary>Raw output</summary> ``` $ git checkout origin/main -- airflow-core/src/airflow/jobs/scheduler_job_runner.py $ uv run --project airflow-core pytest airflow-core/tests/unit/jobs/test_scheduler_job.py -k histogram -q E AssertionError: timing('pool.queued_slots_histogram', 0, tags={'pool_name': 'unbounded_pool'}) call not found FAILED ...::test_emit_pool_metrics_emits_histogram_alongside_gauge FAILED ...::test_emit_pool_metrics_skips_open_slots_histogram_for_unbounded_pool 2 failed, 409 deselected, 1 warning in 5.76s $ git checkout HEAD -- airflow-core/src/airflow/jobs/scheduler_job_runner.py $ uv run --project airflow-core pytest airflow-core/tests/unit/jobs/test_scheduler_job.py -k emit_pool_metrics -q airflow-core/tests/unit/jobs/test_scheduler_job.py::TestSchedulerJob::test_emit_pool_metrics_team_name[with_team] PASSED airflow-core/tests/unit/jobs/test_scheduler_job.py::TestSchedulerJob::test_emit_pool_metrics_team_name[without_team] PASSED airflow-core/tests/unit/jobs/test_scheduler_job.py::TestSchedulerJob::test_emit_pool_metrics_emits_histogram_alongside_gauge PASSED airflow-core/tests/unit/jobs/test_scheduler_job.py::TestSchedulerJob::test_emit_pool_metrics_skips_open_slots_histogram_for_unbounded_pool PASSED 4 passed, 407 deselected, 1 warning in 18.60s ``` </details> Registry sync and type checks are clean (`check-metrics-synced-with-registry`, `mypy-airflow-core`, `prek run --from-ref origin/main --stage pre-commit`). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
