jedcunningham commented on code in PR #31725:
URL: https://github.com/apache/airflow/pull/31725#discussion_r1226933066
##########
airflow/metrics/otel_logger.py:
##########
@@ -100,22 +153,22 @@ def incr(
rate: float = 1,
tags: Attributes = None,
):
- """
+ """counter
Review Comment:
Was this change intentional?
##########
airflow/metrics/otel_logger.py:
##########
@@ -235,6 +304,60 @@ def del_counter(self, name: str, attributes: Attributes =
None) -> None:
if key in self.map.keys():
del self.map[key]
+ def set_gauge_value(self, name: str, value: float, delta: bool, tags:
Attributes):
+ """
+ Overrides the last reading for a Gauge with a new value.
+
+ :param name: The name of the gauge to record.
+ :param value: The new reading to record.
+ :param delta: If True, value is added to the previous reading, else it
overrides.
+ :param tags: Gauge attributes which were used to generate a unique key
to store the counter.
+ :returns: None
+ """
+ key: str = _generate_key_name(name, tags)
+ new_value = value
+ old_value = self.poke_gauge(name, tags)
+ if delta:
+ new_value += old_value
+ # If delta is true, add the new value to the last reading otherwise
overwrite it.
+ self.map[key] = Observation(new_value, tags)
+
+ def _create_gauge(self, name: str, attributes: Attributes = None):
+ """
+ Creates a new Observable Gauge with the provided name and the default
value.
+
+ :param name: The name of the gauge to fetch or create.
+ :param attributes: Gauge attributes, used to generate a unique key to
store the gauge.
+ """
+ otel_safe_name = _get_otel_safe_name(name)
+ key = _generate_key_name(name, attributes)
+
+ gauge = self.meter.create_observable_gauge(
+ name=otel_safe_name,
+ callbacks=[partial(self.read_gauge, _generate_key_name(name,
attributes))],
+ )
+ self.map[key] = Observation(DEFAULT_GAUGE_VALUE, attributes)
+
+ return gauge
+
+ def read_gauge(self, key: str, *args) -> Iterable[Observation]:
+ """Callback for the Observable Gauges, returns the Observation for the
provided key."""
+ yield self.map[key]
+
+ def poke_gauge(self, name: str, attributes: Attributes = None) ->
GaugeValues:
+ """
+ Returns the value of the gauge; creates a new one with the default
value if it did not exist.
+
+ :param name: The name of the gauge to fetch or create.
+ :param attributes: Gauge attributes, used to generate a unique key to
store the gauge.
+ :returns: The integer or float value last recorded for the provided
Gauge name.
+ """
+ key = _generate_key_name(name, attributes)
+ if key not in self.map.keys():
Review Comment:
```suggestion
if key not in self.map:
```
nit
##########
airflow/metrics/otel_logger.py:
##########
@@ -177,7 +252,7 @@ def timer(
tags: Attributes = None,
**kwargs,
) -> TimerProtocol:
- warnings.warn("OpenTelemetry Timers are not yet implemented.")
+ warnings.warn(f"Create timer {stat} : OpenTelemetry Timers are not yet
implemented.")
Review Comment:
```suggestion
warnings.warn(f"Create timer {stat}: OpenTelemetry Timers are not
yet implemented.")
```
Another tiny nit
##########
airflow/metrics/otel_logger.py:
##########
@@ -166,7 +241,7 @@ def timing(
*,
tags: Attributes = None,
) -> None:
- warnings.warn("OpenTelemetry Timers are not yet implemented.")
+ warnings.warn(f"Create timer {stat} : OpenTelemetry Timers are not yet
implemented.")
Review Comment:
```suggestion
warnings.warn(f"Create timer {stat}: OpenTelemetry Timers are not
yet implemented.")
```
tiny nit
--
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]
