xBis7 commented on code in PR #68393:
URL: https://github.com/apache/airflow/pull/68393#discussion_r3420632757
##########
shared/observability/src/airflow_shared/observability/metrics/otel_logger.py:
##########
@@ -435,70 +437,134 @@ 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.
+
+ ``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``.
- 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).
+ 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 = (
+ os.environ.get("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL")
+ or os.environ.get("OTEL_EXPORTER_OTLP_PROTOCOL")
+ or "http/protobuf"
+ )
+ 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)
Review Comment:
It looks good. Out of curiosity, what happens if the exporter is `zipkin` or
`prometheus`? The SDK takes care of the protocol?
Can you add some unit tests for the new logic? I don't know if there are and
I've missed them.
Also, traces should follow the same approach. The code seems pretty similar,
we can probably extract it to a common function.
--
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]
