potiuk commented on code in PR #68393:
URL: https://github.com/apache/airflow/pull/68393#discussion_r3658456358
##########
airflow-core/src/airflow/settings.py:
##########
@@ -816,6 +835,7 @@ def initialize():
import_local_settings()
configure_logging()
configure_otel(conf)
+ _initialize_stats()
Review Comment:
This is the part I'm least comfortable with (see the review body for the
longer version).
`settings.initialize()` runs once, from `airflow/__init__.py:79`, at `import
airflow` — it is not re-run in a forked child. The call sites this PR removes
were all *post-fork* re-inits, and `stats.initialize()` setting `_backend =
None` is what forces the `MeterProvider` and its
`PeriodicExportingMetricReader` thread to be rebuilt in the child. That thread
doesn't survive `fork()`.
Under gunicorn with `"preload_app": True`
(`airflow-core/src/airflow/api_fastapi/gunicorn_app.py:276`) `import airflow`
happens in the master, so if anything touches `Stats` during preload each
worker inherits a provider with a dead exporter thread and nothing left to
reset it. That's the failure mode #64703 fixed, and
`_initialize_api_server_stats()` was added in #68514 for exactly this.
Worth noting too: the fork guard you carefully kept in `configure_otel` —
```python
_metrics_internal._METER_PROVIDER_SET_ONCE._done = False
```
— only does anything if something re-runs the factory in the child. I can't
see what would after this change, which suggests it (and issue #64690 behind
it) becomes unreachable in the scenario its own comment describes.
Centralising the configuration here seems right; it's the per-process reset
that I think needs to stay at each fork boundary, or move to an
`os.register_at_fork` hook.
##########
shared/observability/src/airflow_shared/observability/traces/__init__.py:
##########
@@ -229,15 +231,32 @@ def _get_backcompat_config(conf: ConfigParser) ->
tuple[str | None, Resource | N
def _load_exporter_from_env() -> SpanExporter:
"""
- Load a span exporter using the OTEL_TRACES_EXPORTER env var.
+ Pick a span exporter per the OTel SDK environment-variable spec.
+
+ ``OTEL_TRACES_EXPORTER`` selects the backend (``otlp`` default; ``console``
+ for debugging; ``zipkin`` or custom values are looked up via entry points).
+ For ``otlp``, ``OTEL_EXPORTER_OTLP_TRACES_PROTOCOL`` then
+ ``OTEL_EXPORTER_OTLP_PROTOCOL`` selects the transport: ``http/protobuf``
+ (default) or ``grpc``.
- Mirrors the entry-point mechanism used by the OTEL SDK auto-instrumentation
- configurator. Supported values (from installed packages):
- - ``otlp`` (default) — OTLP/gRPC
- - ``otlp_proto_http`` — OTLP/HTTP
- - ``console`` — stdout (useful for debugging)
+ See:
+
https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#exporter-selection
+
https://opentelemetry.io/docs/specs/otel/protocol/exporter/#specify-protocol
"""
exporter_name = os.environ.get("OTEL_TRACES_EXPORTER", "otlp")
+ if exporter_name == "otlp":
Review Comment:
This changes the default traces transport, and I think it needs a
newsfragment.
On `main`, `OTEL_TRACES_EXPORTER=otlp` (the default) resolves through the
OTel entry point to the **gRPC** span exporter — the docstring being replaced
says so:
```
- ``otlp`` (default) — OTLP/gRPC
- ``otlp_proto_http`` — OTLP/HTTP
```
The new short-circuit makes `otlp` mean `http/protobuf` unless
`OTEL_EXPORTER_OTLP_[TRACES_]PROTOCOL` says otherwise, and your doc change in
this PR says the same:
> `- # Core tracing defaults the exporter to OTLP/gRPC.`
> `+ # Core tracing defaults to OTLP over HTTP/protobuf (port 4318, ...)`
Matching the OTel spec default is defensible, but existing deployments
pointing `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` at a gRPC collector on 4317
without an explicit protocol will silently stop delivering traces.
Per [`AGENTS.md`](https://github.com/apache/airflow/blob/main/AGENTS.md):
> For `airflow-core` (and `chart/`, `dev/mypy/`) **user-facing** changes,
add a newsfragment in that distribution's `newsfragments/` directory.
`shared/` and `task-sdk/` ship inside `airflow-core`, so could you add
`airflow-core/newsfragments/68393.significant.rst` noting the new default and
the one-line `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=grpc` migration?
##########
shared/observability/src/airflow_shared/observability/metrics/otel_logger.py:
##########
@@ -435,70 +437,130 @@ def atexit_register_metrics_flush():
atexit.register(flush_otel_metrics)
-def get_otel_logger(
+def _get_backcompat_config(
Review Comment:
The two deprecation warnings from the removed `get_otel_data_exporter` don't
have a replacement here:
```python
log.warning(
"Both the standard OpenTelemetry environment variables and "
"the Airflow OpenTelemetry configs have been provided. "
"Using the OpenTelemetry environment variables. ..."
)
...
log.warning(
"The Airflow OpenTelemetry configs have been deprecated and will
be removed in the future. ..."
)
```
After this change the conf values are silently ignored when the env vars are
set, and silently bridged when they aren't. All of these keys carry
`version_deprecated: 3.2.0` in `config.yml`, and that runtime warning is the
only heads-up a deployment gets before they're removed — I'd keep both on the
bridging path.
##########
airflow-core/tests/unit/core/test_settings.py:
##########
@@ -533,3 +533,50 @@ def test_early_return_when_all_none(self):
settings.dispose_orm(do_log=False)
mock_close.assert_not_called()
+
+
+class TestInitializeStats:
Review Comment:
These replace `TestInitializeApiServerStats` (3 tests, including
`test_stats_initialized_during_lifespan`),
`test_stats_initialize_called_on_run` (Dag processor) and
`test_stats_initialize_called_on_execute` (triggerer) — but they only assert
that `settings.initialize()` calls `settings._initialize_stats()`.
So nothing asserts any more that the API server / Dag processor / triggerer
/ executor processes actually end up with an initialised `Stats` backend, which
is the property my concern on `settings.py` is about. From [`AGENTS.md` §
Testing Standards](https://github.com/apache/airflow/blob/main/AGENTS.md):
> Every changed or added behaviour must have a test; every test must fail
without the PR's change.
Minor aside on `test_stats_initialized_during_initialize` below: it calls
the real `settings.initialize()` with only
`prepare_syspath_for_config_and_plugins` and `import_local_settings` mocked, so
`configure_orm` / `configure_logging` / `configure_otel` all run for real in a
unit test.
##########
shared/observability/src/airflow_shared/observability/metrics/otel_logger.py:
##########
@@ -435,70 +437,130 @@ def atexit_register_metrics_flush():
atexit.register(flush_otel_metrics)
-def get_otel_logger(
+def _get_backcompat_config(
*,
- host: str | None = None,
- port: int | None = None,
- prefix: str | None = None,
- ssl_active: bool = False,
- conf_interval: float | None = None,
- debug: bool = False,
- service_name: str | None = None,
- metrics_allow_list: str | None = None,
- metrics_block_list: str | None = None,
- stat_name_handler: Callable[[str], str] | None = None,
- statsd_influxdb_enabled: bool = False,
-) -> SafeOtelLogger:
+ host: str | None,
+ port: str | None,
+ ssl_active: bool,
+ service: str | None,
+ interval_ms: str | None,
+) -> tuple[str | None, float | None, Resource | None]:
+ resource = None
+ if service and not os.environ.get("OTEL_SERVICE_NAME") and not
os.environ.get("OTEL_RESOURCE_ATTRIBUTES"):
+ resource = Resource.create(attributes={SERVICE_NAME: service})
+
+ endpoint = None
+ if (
+ host
+ and port
+ and not os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT")
+ and not os.environ.get("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT")
+ ):
+ scheme = "https" if ssl_active else "http"
+ endpoint = f"{scheme}://{_format_url_host(host)}:{port}/v1/metrics"
+
+ parsed_interval_ms: float | None = None
+ if interval_ms and not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL"):
+ try:
+ parsed_interval_ms = float(interval_ms)
+ except (TypeError, ValueError):
+ log.warning("Invalid metrics.otel_interval_milliseconds value: %r;
ignoring.", interval_ms)
+
+ return endpoint, parsed_interval_ms, resource
+
+
+def _load_exporter_from_env() -> MetricExporter:
"""
- Build and return a :class:`SafeOtelLogger` backed by a configured
:class:`MeterProvider`.
+ Pick a metric exporter per the OTel SDK environment-variable spec.
- Histogram instruments (used for ``timing()`` / ``timer()`` metrics) are
aggregated with
-
:class:`~opentelemetry.sdk.metrics.view.ExponentialBucketHistogramAggregation`
- so that bucket boundaries adapt automatically to the observed data range.
This avoids
- the need to hand-tune explicit bucket boundaries for metrics that span
very different
- scales (milliseconds to hours).
+ ``OTEL_METRICS_EXPORTER`` selects the backend (``otlp`` default;
``console``
+ for debugging; ``prometheus`` or custom values are looked up via entry
+ points). For ``otlp``, ``OTEL_EXPORTER_OTLP_METRICS_PROTOCOL`` then
+ ``OTEL_EXPORTER_OTLP_PROTOCOL`` selects the transport: ``http/protobuf``
+ (default) or ``grpc``.
+
+ See:
+
https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#exporter-selection
+
https://opentelemetry.io/docs/specs/otel/protocol/exporter/#specify-protocol
"""
- otel_env_config = load_metrics_env_config()
+ exporter_name = os.environ.get("OTEL_METRICS_EXPORTER", "otlp")
+ if exporter_name == "otlp":
+ protocol =
_resolve_otlp_protocol("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL")
+ if protocol == "http/protobuf":
+ from opentelemetry.exporter.otlp.proto.http.metric_exporter import
OTLPMetricExporter
+
+ return OTLPMetricExporter()
+ if protocol == "grpc":
+ from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import
( # type: ignore[assignment]
+ OTLPMetricExporter,
+ )
- effective_service_name: str = otel_env_config.service_name or service_name
or "airflow"
- effective_prefix: str = prefix or DEFAULT_METRIC_NAME_PREFIX
- resource = Resource.create(attributes={SERVICE_NAME:
effective_service_name})
+ return OTLPMetricExporter()
+ raise ValueError(f"Unsupported OTLP protocol {protocol!r}; expected
'grpc' or 'http/protobuf'.")
+ eps = entry_points(group="opentelemetry_metrics_exporter",
name=exporter_name)
+ ep = next(iter(eps), None)
+ if ep is None:
+ raise RuntimeError(
+ f"No metric exporter found for
OTEL_METRICS_EXPORTER={exporter_name!r}. "
+ f"Available: {[e.name for e in
entry_points(group='opentelemetry_metrics_exporter')]}"
+ )
+ return ep.load()()
- #
https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#periodic-exporting-metricreader
- interval = otel_env_config.interval_ms or conf_interval
- metric_exporter = get_otel_data_exporter(
- otel_env_config=otel_env_config,
+def configure_otel(
+ *,
+ host: str | None = None,
+ port: str | None = None,
+ ssl_active: bool = False,
+ service: str | None = None,
+ interval_ms: str | None = None,
+ debug: bool = False,
+ prefix: str = DEFAULT_METRIC_NAME_PREFIX,
+ allow_list: str | None = None,
+ block_list: str | None = None,
+ stat_name_handler: Callable[[str], str] | None = None,
+ statsd_influxdb_enabled: bool = False,
+) -> SafeOtelLogger:
+ backcompat_endpoint, backcompat_interval_ms, resource =
_get_backcompat_config(
host=host,
port=port,
ssl_active=ssl_active,
+ service=service,
+ interval_ms=interval_ms,
)
- readers = [
+ if backcompat_endpoint:
+ os.environ["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] = backcompat_endpoint
Review Comment:
Nit: these two writes are process-global — they're inherited by every
subprocess started afterwards, and they become the "env wins" input on any
later `configure_otel()` call in the same process.
It mirrors what `traces/configure_otel` already does, so the symmetry is
clearly deliberate — just worth a one-line comment saying so, like the traces
side carries.
##########
airflow-core/src/airflow/observability/metrics/otel_logger.py:
##########
@@ -26,28 +26,16 @@
def get_otel_logger() -> SafeOtelLogger:
- # The config values have been deprecated and therefore,
- # if the user hasn't added them to the config, the default values won't be
used.
- # A fallback is needed to avoid an exception.
- port = None
- if conf.has_option("metrics", "otel_port"):
- port = conf.getint("metrics", "otel_port")
-
- conf_interval = None
- if conf.has_option("metrics", "otel_interval_milliseconds"):
- conf_interval = conf.getfloat("metrics", "otel_interval_milliseconds")
-
- return otel_logger.get_otel_logger(
- host=conf.get("metrics", "otel_host", fallback=None), # ex:
"breeze-otel-collector"
- port=port, # ex: 4318
- prefix=conf.get("metrics", "otel_prefix", fallback=None), # ex:
"airflow"
+ return otel_logger.configure_otel(
+ host=conf.get("metrics", "otel_host", fallback=None),
+ port=conf.get("metrics", "otel_port", fallback=None),
ssl_active=conf.getboolean("metrics", "otel_ssl_active",
fallback=False),
- # PeriodicExportingMetricReader will default to an interval of 60000
millis.
- conf_interval=conf_interval, # ex: 30000
+ service=conf.get("metrics", "otel_service", fallback=None),
+ interval_ms=conf.get("metrics", "otel_interval_milliseconds",
fallback=None),
debug=conf.getboolean("metrics", "otel_debugging_on", fallback=False),
- service_name=conf.get("metrics", "otel_service", fallback=None),
- metrics_allow_list=conf.get("metrics", "metrics_allow_list",
fallback=None),
- metrics_block_list=conf.get("metrics", "metrics_block_list",
fallback=None),
+ prefix=conf.get("metrics", "otel_prefix", fallback="airflow"),
Review Comment:
`fallback="airflow"` hardcodes the value of `DEFAULT_METRIC_NAME_PREFIX`,
which is exported from the same module you're calling into
(`shared/observability/.../metrics/otel_logger.py`) — and it's duplicated in
the task-SDK shim too.
There's also a small behaviour change hiding here: `configure_otel` now uses
`prefix` directly, where `get_otel_logger` did `prefix or
DEFAULT_METRIC_NAME_PREFIX`. So an explicitly-empty `metrics.otel_prefix` used
to fall back to `airflow` and now strips the prefix from every metric name.
Importing the constant and keeping `fallback=None` would take care of both.
##########
task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py:
##########
@@ -26,28 +26,16 @@
def get_otel_logger() -> SafeOtelLogger:
- # The config values have been deprecated and therefore,
- # if the user hasn't added them to the config, the default values won't be
used.
- # A fallback is needed to avoid an exception.
- port = None
- if conf.has_option("metrics", "otel_port"):
- port = conf.getint("metrics", "otel_port")
-
- conf_interval = None
- if conf.has_option("metrics", "otel_interval_milliseconds"):
- conf_interval = conf.getfloat("metrics", "otel_interval_milliseconds")
-
- return otel_logger.get_otel_logger(
- host=conf.get("metrics", "otel_host", fallback=None), # ex:
"breeze-otel-collector"
- port=port, # ex: 4318
- prefix=conf.get("metrics", "otel_prefix", fallback=None), # ex:
"airflow"
+ return otel_logger.configure_otel(
+ host=conf.get("metrics", "otel_host", fallback=None),
+ port=conf.get("metrics", "otel_port", fallback=None),
ssl_active=conf.getboolean("metrics", "otel_ssl_active",
fallback=False),
- # PeriodicExportingMetricReader will default to an interval of 60000
millis.
- conf_interval=conf_interval, # ex: 30000
+ service=conf.get("metrics", "otel_service", fallback=None),
+ interval_ms=conf.get("metrics", "otel_interval_milliseconds",
fallback=None),
debug=conf.getboolean("metrics", "otel_debugging_on", fallback=False),
- service_name=conf.get("metrics", "otel_service", fallback=None),
- metrics_allow_list=conf.get("metrics", "metrics_allow_list",
fallback=None),
- metrics_block_list=conf.get("metrics", "metrics_block_list",
fallback=None),
+ prefix=conf.get("metrics", "otel_prefix", fallback="airflow"),
Review Comment:
Same as the `airflow-core` shim: `fallback="airflow"` duplicates
`DEFAULT_METRIC_NAME_PREFIX`, and an explicitly-empty `metrics.otel_prefix` now
strips the prefix instead of falling back to it. Importing the constant and
keeping `fallback=None` covers both.
--
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]
