ferruzzi commented on code in PR #56150:
URL: https://github.com/apache/airflow/pull/56150#discussion_r2392953273
##########
airflow-core/src/airflow/metrics/otel_logger.py:
##########
@@ -371,25 +371,37 @@ def set_gauge_value(self, name: str, value: int | float,
delta: bool, tags: Attr
def get_otel_logger(cls) -> SafeOtelLogger:
+ # Check Airflow config.
host = conf.get("metrics", "otel_host") # ex: "breeze-otel-collector"
port = conf.getint("metrics", "otel_port") # ex: 4318
- prefix = conf.get("metrics", "otel_prefix") # ex: "airflow"
- ssl_active = conf.getboolean("metrics", "otel_ssl_active")
- # PeriodicExportingMetricReader will default to an interval of 60000
millis.
- interval = conf.getint("metrics", "otel_interval_milliseconds",
fallback=None) # ex: 30000
- debug = conf.getboolean("metrics", "otel_debugging_on")
- service_name = conf.get("metrics", "otel_service")
- resource = Resource.create(attributes={HOST_NAME: get_hostname(),
SERVICE_NAME: service_name})
+ # If the host or the port hasn't been provided, then check the regular
OTel env vars.
+ if host != "-" and port != "0":
+ ssl_active = conf.getboolean("metrics", "otel_ssl_active")
+ # PeriodicExportingMetricReader will default to an interval of 60000
millis.
+ interval_ms = conf.getint("metrics", "otel_interval_milliseconds",
fallback=None) # ex: 30000
+ debug = conf.getboolean("metrics", "otel_debugging_on")
+ service_name = conf.get("metrics", "otel_service")
+
+ protocol = "https" if ssl_active else "http"
+ endpoint = f"{protocol}://{host}:{port}/v1/metrics"
+ else:
+ otel_config = load_metrics_config()
+ # PeriodicExportingMetricReader will default to an interval of 60000
millis.
Review Comment:
I think we can drop the comment on this one since it's pulling it from the
`load_metrics_config()` result, but I'll leave that up to you, feel free to
resolve this instead.
##########
airflow-core/src/airflow/traces/otel_tracer.py:
##########
@@ -329,11 +337,19 @@ def get_otel_tracer(cls, use_simple_processor: bool =
False) -> OtelTrace:
"""Get OTEL tracer from airflow configuration."""
host = conf.get("traces", "otel_host")
port = conf.getint("traces", "otel_port")
- ssl_active = conf.getboolean("traces", "otel_ssl_active")
+
+ # If the host or the port hasn't been provided, then check the regular
OTel env vars.
+ if host != "-" and port != "0":
Review Comment:
Same as above; comment says OR but the code uses AND, double check that is
doing what you intended.
##########
airflow-core/src/airflow/traces/otel_tracer.py:
##########
@@ -138,9 +144,10 @@ def start_span(
links=None,
start_time=None,
):
- """Start a span; if service_name is not given, otel_service is used."""
+ """Start a span."""
if component is None:
- component = self.otel_service
+ # Common practice is to use the module name.
+ component = __name__
Review Comment:
here and below: Is this going to be a breaking change?
Also, a nitpick you can feel free to ignore since I'm here:
```component = component or __name__```
##########
airflow-core/docs/administration-and-deployment/logging-monitoring/metrics.rst:
##########
@@ -69,11 +69,48 @@ Add the following lines to your configuration file e.g.
``airflow.cfg``
[metrics]
otel_on = True
- otel_host = localhost
- otel_port = 8889
otel_prefix = airflow
- otel_interval_milliseconds = 30000 # The interval between exports,
defaults to 60000
- otel_ssl_active = False
+
+Configure the SDK, by exporting the regular OpenTelemetry variables to your
environment
+
+.. code-block:: ini
+
+ - exporter
+ |_ values: 'otlp', 'console'
+ |_ default: 'otlp'
+ OTEL_METRICS_EXPORTER
+
+ - export protocol
+ |_ values: 'grpc', 'http/protobuf'
+ |_ default: 'grpc'
+ OTEL_EXPORTER_OTLP_PROTOCOL
+
+ - endpoint
+ |_ example for grpc protocol: 'http://localhost:4317'
+ |_ example for http protocol: 'http://localhost:4318/v1/metrics'
+ |_ if SSL is enabled, use 'https' instead of 'http'
+ OTEL_EXPORTER_OTLP_ENDPOINT
+ or
+ OTEL_EXPORTER_OTLP_METRICS_ENDPOINT
+
+ - service name
+ |_ default: 'Airflow'
+ OTEL_SERVICE_NAME
+
+ - resource attributes
+ |_ values: 'key1=value1,key2=value2,...'
+ |_ example: 'service.name=my-service,service.version=1.0.0'
+ OTEL_RESOURCE_ATTRIBUTES
+
+ - list of headers to apply to all outgoing metrics
+ |_ values: 'key1=value1,key2=value2,...'
+ |_ example: 'api-key=key,other-config-value=value'
+ OTEL_EXPORTER_OTLP_HEADERS
+
+ - export interval
+ |_ values: integer or float
+ |_ default: 60000
Review Comment:
Just double checking that this change was intentional, the old value here
was 30000
> otel_interval_milliseconds = 30000 # The interval between exports,
defaults to 60000
##########
airflow-core/tests/unit/core/test_otel_tracer.py:
##########
@@ -40,11 +39,15 @@ def name():
class TestOtelTrace:
Review Comment:
Nice use of the env_vars decorator, it really cleans up the tests
##########
airflow-core/src/airflow/utils/otel_config.py:
##########
@@ -0,0 +1,189 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from enum import Enum
+from functools import lru_cache
+
+import structlog
+
+log = structlog.getLogger(__name__)
+
+
+def _parse_kv_str_to_dict(str_var: str) -> dict[str, str]:
+ """
+ Convert a string of key-value pairs to a dictionary.
+
+ Environment variables like 'OTEL_RESOURCE_ATTRIBUTES' or
'OTEL_EXPORTER_OTLP_HEADERS'
+ accept values with the format "key1=value1,key2=value2,..."
+ """
+ configs = {}
+ if str_var:
+ for pair in str_var.split(","):
+ if "=" in pair:
+ k, v = pair.split("=", 1)
+ configs[k.strip()] = v.strip()
+ return configs
+
+
+class OtelDataType(str, Enum):
+ """Enum with the different telemetry data types."""
+
+ TRACES = "traces"
+ METRICS = "metrics"
+ LOGS = "logs"
+
+
+@dataclass(frozen=True)
+class OtelConfig:
+ """Immutable class for holding and validating OTel config environment
variables."""
+
+ data_type: OtelDataType # traces | metrics
+ endpoint: str # url
+ protocol: str # "grpc" or "http/protobuf"
+ exporter: str # OTEL_TRACES_EXPORTER | OTEL_METRICS_EXPORTER
+ service_name: str # default "Airflow"
+ headers_kv_str: str
+ headers: dict[str, str]
+ resource_attributes_kv_str: str
+ resource_attributes: dict[str, str]
+ interval_ms: int
+ validate: bool # true by default
+
+ def __post_init__(self):
+ """Validate the environment variables where necessary."""
+ if self.validate is False:
+ return
+
+ endpoint_type_specific = (
+ "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"
+ if self.data_type == OtelDataType.TRACES
+ else "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"
+ )
+
+ if not self.endpoint:
+ raise ValueError(
+ f"Missing required environment variable:
'OTEL_EXPORTER_OTLP_ENDPOINT' or {endpoint_type_specific}"
+ )
+
+ stripped_protocol = (self.protocol or
"").strip().strip('"').strip("'").lower()
Review Comment:
Feel free to ignore, but you can combine those strips:
```
stripped_protocol = (self.protocol or "").strip('" \'').lower()
```
but maybe that is harder to read.
##########
airflow-core/src/airflow/utils/otel_config.py:
##########
@@ -0,0 +1,181 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import logging
+import os
+from dataclasses import dataclass
+from enum import Enum
+from functools import lru_cache
+
+log = logging.getLogger(__name__)
+
+
+def _parse_kv_str_to_dict(str_var: str) -> dict[str, str]:
Review Comment:
I'm actually shocked we don't already have a string-to-dict helper in
airflow/utils. I think we used to have one for env_var parsing, but can't find
it.
##########
airflow-core/src/airflow/utils/otel_config.py:
##########
@@ -0,0 +1,189 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from enum import Enum
+from functools import lru_cache
+
+import structlog
+
+log = structlog.getLogger(__name__)
+
+
+def _parse_kv_str_to_dict(str_var: str) -> dict[str, str]:
+ """
+ Convert a string of key-value pairs to a dictionary.
+
+ Environment variables like 'OTEL_RESOURCE_ATTRIBUTES' or
'OTEL_EXPORTER_OTLP_HEADERS'
+ accept values with the format "key1=value1,key2=value2,..."
+ """
+ configs = {}
+ if str_var:
+ for pair in str_var.split(","):
+ if "=" in pair:
+ k, v = pair.split("=", 1)
+ configs[k.strip()] = v.strip()
+ return configs
+
+
+class OtelDataType(str, Enum):
+ """Enum with the different telemetry data types."""
+
+ TRACES = "traces"
+ METRICS = "metrics"
+ LOGS = "logs"
+
+
+@dataclass(frozen=True)
+class OtelConfig:
+ """Immutable class for holding and validating OTel config environment
variables."""
+
+ data_type: OtelDataType # traces | metrics
+ endpoint: str # url
+ protocol: str # "grpc" or "http/protobuf"
+ exporter: str # OTEL_TRACES_EXPORTER | OTEL_METRICS_EXPORTER
+ service_name: str # default "Airflow"
+ headers_kv_str: str
+ headers: dict[str, str]
+ resource_attributes_kv_str: str
+ resource_attributes: dict[str, str]
+ interval_ms: int
+ validate: bool # true by default
+
+ def __post_init__(self):
+ """Validate the environment variables where necessary."""
+ if self.validate is False:
+ return
+
+ endpoint_type_specific = (
+ "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"
+ if self.data_type == OtelDataType.TRACES
+ else "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"
+ )
Review Comment:
```suggestion
endpoint_type_specific =
f"OTEL_EXPORTER_OTLP_{self.data_type.name}_ENDPOINT"
```
##########
airflow-core/src/airflow/utils/otel_config.py:
##########
@@ -0,0 +1,189 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from enum import Enum
+from functools import lru_cache
+
+import structlog
+
+log = structlog.getLogger(__name__)
+
+
+def _parse_kv_str_to_dict(str_var: str) -> dict[str, str]:
+ """
+ Convert a string of key-value pairs to a dictionary.
+
+ Environment variables like 'OTEL_RESOURCE_ATTRIBUTES' or
'OTEL_EXPORTER_OTLP_HEADERS'
+ accept values with the format "key1=value1,key2=value2,..."
+ """
+ configs = {}
+ if str_var:
+ for pair in str_var.split(","):
+ if "=" in pair:
+ k, v = pair.split("=", 1)
+ configs[k.strip()] = v.strip()
+ return configs
+
+
+class OtelDataType(str, Enum):
+ """Enum with the different telemetry data types."""
+
+ TRACES = "traces"
+ METRICS = "metrics"
+ LOGS = "logs"
+
+
+@dataclass(frozen=True)
+class OtelConfig:
+ """Immutable class for holding and validating OTel config environment
variables."""
+
+ data_type: OtelDataType # traces | metrics
+ endpoint: str # url
+ protocol: str # "grpc" or "http/protobuf"
+ exporter: str # OTEL_TRACES_EXPORTER | OTEL_METRICS_EXPORTER
+ service_name: str # default "Airflow"
+ headers_kv_str: str
+ headers: dict[str, str]
+ resource_attributes_kv_str: str
+ resource_attributes: dict[str, str]
+ interval_ms: int
+ validate: bool # true by default
Review Comment:
Consistency nitpick here too: `service_name: str # default "Airflow"` or
`validate: bool # true by default`, pick one. I don't care which.
##########
airflow-core/src/airflow/utils/otel_config.py:
##########
@@ -0,0 +1,189 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from enum import Enum
+from functools import lru_cache
+
+import structlog
+
+log = structlog.getLogger(__name__)
+
+
+def _parse_kv_str_to_dict(str_var: str) -> dict[str, str]:
+ """
+ Convert a string of key-value pairs to a dictionary.
+
+ Environment variables like 'OTEL_RESOURCE_ATTRIBUTES' or
'OTEL_EXPORTER_OTLP_HEADERS'
+ accept values with the format "key1=value1,key2=value2,..."
+ """
+ configs = {}
+ if str_var:
+ for pair in str_var.split(","):
+ if "=" in pair:
+ k, v = pair.split("=", 1)
+ configs[k.strip()] = v.strip()
+ return configs
+
+
+class OtelDataType(str, Enum):
+ """Enum with the different telemetry data types."""
+
+ TRACES = "traces"
+ METRICS = "metrics"
+ LOGS = "logs"
+
+
+@dataclass(frozen=True)
+class OtelConfig:
+ """Immutable class for holding and validating OTel config environment
variables."""
+
+ data_type: OtelDataType # traces | metrics
+ endpoint: str # url
+ protocol: str # "grpc" or "http/protobuf"
+ exporter: str # OTEL_TRACES_EXPORTER | OTEL_METRICS_EXPORTER
+ service_name: str # default "Airflow"
+ headers_kv_str: str
+ headers: dict[str, str]
+ resource_attributes_kv_str: str
+ resource_attributes: dict[str, str]
+ interval_ms: int
+ validate: bool # true by default
+
+ def __post_init__(self):
+ """Validate the environment variables where necessary."""
+ if self.validate is False:
+ return
+
+ endpoint_type_specific = (
+ "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"
+ if self.data_type == OtelDataType.TRACES
+ else "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"
+ )
+
+ if not self.endpoint:
+ raise ValueError(
+ f"Missing required environment variable:
'OTEL_EXPORTER_OTLP_ENDPOINT' or {endpoint_type_specific}"
+ )
+
+ stripped_protocol = (self.protocol or
"").strip().strip('"').strip("'").lower()
Review Comment:
You might also consider reversing the logic. I think if you check `if
"grpc" in self.protocol or "http/protobuf" in self.protocol:` then you don't
need to bother with the stripping at all, right?
so something like:
```
self.protocol = self.protocol.lower()
if "grpc" not in self.protocol and "http/protobuf" not in self.protocol:
raise ...
if "http/protobuf" in self.protocol:
...
##########
airflow-core/src/airflow/metrics/otel_logger.py:
##########
@@ -371,25 +371,37 @@ def set_gauge_value(self, name: str, value: int | float,
delta: bool, tags: Attr
def get_otel_logger(cls) -> SafeOtelLogger:
+ # Check Airflow config.
host = conf.get("metrics", "otel_host") # ex: "breeze-otel-collector"
port = conf.getint("metrics", "otel_port") # ex: 4318
- prefix = conf.get("metrics", "otel_prefix") # ex: "airflow"
- ssl_active = conf.getboolean("metrics", "otel_ssl_active")
- # PeriodicExportingMetricReader will default to an interval of 60000
millis.
- interval = conf.getint("metrics", "otel_interval_milliseconds",
fallback=None) # ex: 30000
- debug = conf.getboolean("metrics", "otel_debugging_on")
- service_name = conf.get("metrics", "otel_service")
- resource = Resource.create(attributes={HOST_NAME: get_hostname(),
SERVICE_NAME: service_name})
+ # If the host or the port hasn't been provided, then check the regular
OTel env vars.
+ if host != "-" and port != "0":
Review Comment:
Comment says OR, logic says AND. Please confirm.
##########
scripts/ci/docker-compose/integration-otel.yml:
##########
@@ -68,14 +68,15 @@ services:
airflow:
environment:
- INTEGRATION_OTEL=true
Review Comment:
I just want to verify that you've manually tested this with `breeze
start-airflow --integration otel` and it still works?
##########
airflow-core/src/airflow/utils/otel_config.py:
##########
@@ -0,0 +1,189 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from enum import Enum
+from functools import lru_cache
+
+import structlog
+
+log = structlog.getLogger(__name__)
+
+
+def _parse_kv_str_to_dict(str_var: str) -> dict[str, str]:
+ """
+ Convert a string of key-value pairs to a dictionary.
+
+ Environment variables like 'OTEL_RESOURCE_ATTRIBUTES' or
'OTEL_EXPORTER_OTLP_HEADERS'
+ accept values with the format "key1=value1,key2=value2,..."
+ """
+ configs = {}
+ if str_var:
+ for pair in str_var.split(","):
+ if "=" in pair:
+ k, v = pair.split("=", 1)
+ configs[k.strip()] = v.strip()
+ return configs
+
+
+class OtelDataType(str, Enum):
+ """Enum with the different telemetry data types."""
+
+ TRACES = "traces"
+ METRICS = "metrics"
+ LOGS = "logs"
+
+
+@dataclass(frozen=True)
+class OtelConfig:
+ """Immutable class for holding and validating OTel config environment
variables."""
+
+ data_type: OtelDataType # traces | metrics
Review Comment:
Technically this can be 'logs' as well, right? Should we add that to the
comment, or remove it from the enum, or leave it? We don't currently support
OTel Logs, right? Maybe we should remove it for now?
##########
airflow-core/src/airflow/utils/otel_config.py:
##########
@@ -0,0 +1,181 @@
+#
Review Comment:
Approved the other PR, pending Ash reviews and signs off on the SDK question.
##########
airflow-core/src/airflow/utils/otel_config.py:
##########
@@ -0,0 +1,189 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from enum import Enum
+from functools import lru_cache
+
+import structlog
+
+log = structlog.getLogger(__name__)
+
+
+def _parse_kv_str_to_dict(str_var: str) -> dict[str, str]:
+ """
+ Convert a string of key-value pairs to a dictionary.
+
+ Environment variables like 'OTEL_RESOURCE_ATTRIBUTES' or
'OTEL_EXPORTER_OTLP_HEADERS'
+ accept values with the format "key1=value1,key2=value2,..."
+ """
+ configs = {}
+ if str_var:
+ for pair in str_var.split(","):
+ if "=" in pair:
+ k, v = pair.split("=", 1)
+ configs[k.strip()] = v.strip()
+ return configs
+
+
+class OtelDataType(str, Enum):
+ """Enum with the different telemetry data types."""
+
+ TRACES = "traces"
+ METRICS = "metrics"
+ LOGS = "logs"
+
+
+@dataclass(frozen=True)
+class OtelConfig:
+ """Immutable class for holding and validating OTel config environment
variables."""
+
+ data_type: OtelDataType # traces | metrics
+ endpoint: str # url
+ protocol: str # "grpc" or "http/protobuf"
Review Comment:
Consistency nitpick
```suggestion
protocol: str # "grpc" | "http/protobuf"
```
--
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]
